Refactor find_one_await_thumbnail invocation

Author: carlos-villavicencio-adsk

tests/base.py

@@ -3,6 +3,7 @@
 import os
 import random
 import re
+import time
 import unittest
 
 from . import mock
@@ -26,6 +27,11 @@ def skip(f):
         return lambda self: None
 
 
+THUMBNAIL_MAX_ATTEMPTS = 30
+THUMBNAIL_RETRY_INTERVAL = 10
+TRANSIENT_IMAGE_PATH = "images/status/transient"
+
+
 class TestBase(unittest.TestCase):
     '''Base class for tests.
 
@@ -366,6 +372,20 @@ def gen_entity(self, entity_type, **kwargs):
             rv = self.sg.delete(entity_type, entity["id"])
             assert rv == True
 
+    def find_one_await_thumbnail(self, entity_type, filters, fields=["image"], thumbnail_field_name="image", **kwargs):
+        attempts = 0
+        result = self.sg.find_one(entity_type, filters, fields=fields, **kwargs)
+        while attempts < THUMBNAIL_MAX_ATTEMPTS:
+            result = self.sg.find_one(entity_type, filters, fields=fields, **kwargs)
+            if TRANSIENT_IMAGE_PATH in result.get(thumbnail_field_name, ""):
+                return result
+            
+            time.sleep(THUMBNAIL_RETRY_INTERVAL)
+            attempts += 1
+        else:
+            if self.config.jenkins:
+                self.skipTest("Jenkins test timed out waiting for thumbnail")
+
 
 class HumanUserAuthLiveTestBase(LiveTestBase):
     '''

tests/test_api.py

@@ -40,10 +40,6 @@
 
 from . import base
 
-THUMBNAIL_MAX_ATTEMPTS = 30
-THUMBNAIL_RETRY_INTERAL = 10
-TRANSIENT_IMAGE_PATH = "images/status/transient"
-
 
 class TestShotgunApi(base.LiveTestBase):
     def setUp(self):
@@ -383,11 +379,10 @@ def test_upload_thumbnail_in_create(self):
         data = {'image': path, 'code': 'Test Version',
                 'project': self.project}
         new_version = self.sg.create("Version", data, return_fields=['image'])
-        new_version = find_one_await_thumbnail(
-            self.sg,
+        new_version = self.find_one_await_thumbnail(
             "Version",
             [["id", "is", new_version["id"]]],
-            fields=["image", "project", "type", "id"]
+            fields=["image", "project", "type", "id"],
         )
 
         self.assertTrue(new_version is not None)
@@ -434,7 +429,9 @@ def test_upload_thumbnail_for_version(self):
 
         # check result on version
         version_with_thumbnail = self.sg.find_one('Version', [['id', 'is', self.version['id']]])
-        version_with_thumbnail = find_one_await_thumbnail(self.sg, 'Version', [['id', 'is', self.version['id']]])
+        version_with_thumbnail = self.find_one_await_thumbnail(
+            "Version", [["id", "is", self.version["id"]]]
+        )
 
         self.assertEqual(version_with_thumbnail.get('type'), 'Version')
         self.assertEqual(version_with_thumbnail.get('id'), self.version['id'])
@@ -461,7 +458,9 @@ def test_upload_thumbnail_for_task(self):
 
         # check result on version
         task_with_thumbnail = self.sg.find_one('Task', [['id', 'is', self.task['id']]])
-        task_with_thumbnail = find_one_await_thumbnail(self.sg, 'Task', [['id', 'is', self.task['id']]])
+        task_with_thumbnail = self.find_one_await_thumbnail(
+            "Task", [["id", "is", self.task["id"]]]
+        )
 
         self.assertEqual(task_with_thumbnail.get('type'), 'Task')
         self.assertEqual(task_with_thumbnail.get('id'), self.task['id'])
@@ -557,12 +556,11 @@ def test_linked_thumbnail_url(self):
 
         thumb_id = self.sg.upload_thumbnail("Project", self.version['project']['id'], path)
 
-        response_version_with_project = find_one_await_thumbnail(
-            self.sg,
+        response_version_with_project = self.find_one_await_thumbnail(
             "Version",
             [["id", "is", self.version["id"]]],
             fields=["id", "code", "project.Project.image"],
-            thumbnail_field_name="project.Project.image"
+            thumbnail_field_name="project.Project.image",
         )
 
         if self.sg.server_caps.version and self.sg.server_caps.version >= (3, 3, 0):
@@ -597,30 +595,28 @@ def share_thumbnail_retry(*args, **kwargs):
             # the thumbnail to finish processing.
             thumbnail_id = None
             attempts = 0
-            while attempts < THUMBNAIL_MAX_ATTEMPTS and thumbnail_id is None:
+            while attempts < base.THUMBNAIL_MAX_ATTEMPTS and thumbnail_id is None:
                 try:
                     thumbnail_id = self.sg.share_thumbnail(*args, **kwargs)
                     attempts += 1
                 except shotgun_api3.ShotgunError:
-                    time.sleep(THUMBNAIL_RETRY_INTERAL)
+                    time.sleep(base.THUMBNAIL_RETRY_INTERVAL)
             return thumbnail_id
 
         this_dir, _ = os.path.split(__file__)
         path = os.path.abspath(os.path.expanduser(os.path.join(this_dir, "sg_logo.jpg")))
 
         # upload thumbnail to first entity and share it with the rest
         share_thumbnail_retry([self.version, self.shot], thumbnail_path=path)
-        response_version_thumbnail = find_one_await_thumbnail(
-            self.sg,
+        response_version_thumbnail = self.find_one_await_thumbnail(
             'Version',
             [['id', 'is', self.version['id']]],
-            fields=['id', 'code', 'image']
+            fields=['id', 'code', 'image'],
         )
-        response_shot_thumbnail = find_one_await_thumbnail(
-            self.sg,
+        response_shot_thumbnail = self.find_one_await_thumbnail(
             'Shot',
             [['id', 'is', self.shot['id']]],
-            fields=['id', 'code', 'image']
+            fields=['id', 'code', 'image'],
         )
 
         shot_url = urllib.parse.urlparse(response_shot_thumbnail.get('image'))
@@ -632,23 +628,20 @@ def share_thumbnail_retry(*args, **kwargs):
         # share thumbnail from source entity with entities
         self.sg.upload_thumbnail("Version", self.version['id'], path)
         share_thumbnail_retry([self.asset, self.shot], source_entity=self.version)
-        response_version_thumbnail = find_one_await_thumbnail(
-            self.sg,
+        response_version_thumbnail = self.find_one_await_thumbnail(
             'Version',
             [['id', 'is', self.version['id']]],
-            fields=['id', 'code', 'image']
+            fields=['id', 'code', 'image'],
         )
-        response_shot_thumbnail = find_one_await_thumbnail(
-            self.sg,
+        response_shot_thumbnail = self.find_one_await_thumbnail(
             'Shot',
             [['id', 'is', self.shot['id']]],
-            fields=['id', 'code', 'image']
+            fields=['id', 'code', 'image'],
         )
-        response_asset_thumbnail = find_one_await_thumbnail(
-            self.sg,
+        response_asset_thumbnail = self.find_one_await_thumbnail(
             'Asset',
             [['id', 'is', self.asset['id']]],
-            fields=['id', 'code', 'image']
+            fields=['id', 'code', 'image'],
         )
 
         shot_url = urllib.parse.urlparse(response_shot_thumbnail.get('image'))
@@ -856,7 +849,6 @@ def test_work_schedule(self):
         self.assertRaises(shotgun_api3.ShotgunError, self.sg.work_schedule_read,
                           start_date_obj, end_date_obj, project, user)
 
-
         resp = self.sg.work_schedule_update('2012-01-02', False, 'Studio Holiday')
         expected = {
             'date': '2012-01-02',
@@ -1007,7 +999,7 @@ def test_set_date(self):
 
     def test_set_date_time(self):
         if self.config.jenkins:
-            self.skipTest("Skipping test on Jenkins. locked_until not updating.")
+            self.skipTest("Jenkins. locked_until not updating.")
         entity = 'HumanUser'
         entity_id = self.human_user['id']
         field_name = 'locked_until'
@@ -1202,7 +1194,7 @@ def setUp(self):
 
     def test_convert_to_utc(self):
         if self.config.jenkins:
-            self.skipTest("Skipping test on Jenkins. locked_until not updating.")
+            self.skipTest("Jenkins. locked_until not updating.")
         sg_utc = shotgun_api3.Shotgun(self.config.server_url,
                                       http_proxy=self.config.http_proxy,
                                       convert_datetimes_to_utc=True,
@@ -1212,7 +1204,7 @@ def test_convert_to_utc(self):
 
     def test_no_convert_to_utc(self):
         if self.config.jenkins:
-            self.skipTest("Skipping test on Jenkins. locked_until not updating.")
+            self.skipTest("Jenkins. locked_until not updating.")
         sg_no_utc = shotgun_api3.Shotgun(self.config.server_url,
                                          http_proxy=self.config.http_proxy,
                                          convert_datetimes_to_utc=False,
@@ -2164,7 +2156,7 @@ def test_human_user_sudo_auth_fails(self):
         Request fails on server because user has no permission to Sudo.
         """
         if self.config.jenkins:
-            self.skipTest("Skipping test on Jenkins. locked_until not updating.")
+            self.skipTest("Jenkins. locked_until not updating.")
 
         if not self.sg.server_caps.version or self.sg.server_caps.version < (5, 3, 12):
             return
@@ -2221,7 +2213,10 @@ def test_humanuser_upload_thumbnail_for_version(self):
         self.assertTrue(isinstance(thumb_id, int))
 
         # check result on version
-        version_with_thumbnail = find_one_await_thumbnail(self.sg, 'Version', [['id', 'is', self.version['id']]])
+        version_with_thumbnail = self.find_one_await_thumbnail(
+            "Version",
+            [["id", "is", self.version["id"]]],
+        )
 
         self.assertEqual(version_with_thumbnail.get('type'), 'Version')
         self.assertEqual(version_with_thumbnail.get('id'), self.version['id'])
@@ -2278,7 +2273,10 @@ def test_humanuser_upload_thumbnail_for_version(self):
             self.assertTrue(isinstance(thumb_id, int))
 
             # check result on version
-            version_with_thumbnail = find_one_await_thumbnail(self.sg, 'Version', [['id', 'is', self.version['id']]])
+            version_with_thumbnail = self.find_one_await_thumbnail(
+                "Version",
+                [["id", "is", self.version["id"]]],
+            )
 
             self.assertEqual(version_with_thumbnail.get('type'), 'Version')
             self.assertEqual(version_with_thumbnail.get('id'), self.version['id'])
@@ -3040,19 +3038,5 @@ def _get_path(url):
         return url.path
 
 
-def find_one_await_thumbnail(sg, entity_type, filters, fields=["image"], thumbnail_field_name="image", **kwargs):
-    attempts = 0
-    result = sg.find_one(entity_type, filters, fields=fields, **kwargs)
-    while (
-        attempts < THUMBNAIL_MAX_ATTEMPTS
-        and result[thumbnail_field_name]
-        and TRANSIENT_IMAGE_PATH in result[thumbnail_field_name]
-    ):
-        time.sleep(THUMBNAIL_RETRY_INTERAL)
-        result = sg.find_one(entity_type, filters, fields=fields, **kwargs)
-        attempts += 1
-    return result
-
-
 if __name__ == '__main__':
     unittest.main()