OP-3426 - DL support for Maya automatic tests

WIP

tests/integration/hosts/maya/lib.py

@@ -2,10 +2,14 @@ import os
 import pytest
 import shutil
 
-from tests.lib.testing_classes import HostFixtures
+from tests.lib.testing_classes import (
+    HostFixtures,
+    PublishTest,
+    DeadlinePublishTest
+)
 
 
-class MayaTestClass(HostFixtures):
+class MayaHostFixtures(HostFixtures):
     @pytest.fixture(scope="module")
     def last_workfile_path(self, download_test_data, output_folder_url):
         """Get last_workfile_path from source data.
@@ -39,3 +43,11 @@ class MayaTestClass(HostFixtures):
                                    "{}{}{}".format(startup_path,
                                                    os.pathsep,
                                                    original_pythonpath))
+
+
+class MayaLocalPublishTestClass(MayaHostFixtures, PublishTest):
+    """Testing class for local publishes."""
+
+
+class MayaDeadlinePublishTestClass(MayaHostFixtures, DeadlinePublishTest):
+    """Testing class for Deadline publishes."""

tests/integration/hosts/maya/test_deadline_publish_in_maya.py

@@ -0,0 +1,70 @@
+from tests.integration.hosts.maya.lib import MayaDeadlinePublishTestClass
+
+
+class TestDeadlinePublishInMaya(MayaDeadlinePublishTestClass):
+    """Basic test case for publishing in Maya
+
+        Shouldnt be running standalone only via 'runtests' pype command! (??)
+
+        Uses generic TestCase to prepare fixtures for test data, testing DBs,
+        env vars.
+
+        Always pulls and uses test data from GDrive!
+
+        Opens Maya, runs publish on prepared workile.
+
+        Then checks content of DB (if subset, version, representations were
+        created.
+        Checks tmp folder if all expected files were published.
+
+        How to run:
+        (in cmd with activated {OPENPYPE_ROOT}/.venv)
+        {OPENPYPE_ROOT}/.venv/Scripts/python.exe {OPENPYPE_ROOT}/start.py runtests ../tests/integration/hosts/maya  # noqa: E501
+
+    """
+    PERSIST = True
+
+    TEST_FILES = [
+        ("1BTSIIULJTuDc8VvXseuiJV_fL6-Bu7FP", "test_maya_publish.zip", "")
+    ]
+
+    APP = "maya"
+    # keep empty to locate latest installed variant or explicit
+    APP_VARIANT = ""
+
+    TIMEOUT = 120  # publish timeout
+
+    def test_db_asserts(self, dbcon, publish_finished):
+        """Host and input data dependent expected results in DB."""
+        print("test_db_asserts")
+        assert 5 == dbcon.count_documents({"type": "version"}), \
+            "Not expected no of versions"
+
+        assert 0 == dbcon.count_documents({"type": "version",
+                                           "name": {"$ne": 1}}), \
+            "Only versions with 1 expected"
+
+        assert 1 == dbcon.count_documents({"type": "subset",
+                                           "name": "modelMain"}), \
+            "modelMain subset must be present"
+
+        assert 1 == dbcon.count_documents({"type": "subset",
+                                           "name": "workfileTest_task"}), \
+            "workfileTest_task subset must be present"
+
+        assert 11 == dbcon.count_documents({"type": "representation"}), \
+            "Not expected no of representations"
+
+        assert 2 == dbcon.count_documents({"type": "representation",
+                                           "context.subset": "modelMain",
+                                           "context.ext": "abc"}), \
+            "Not expected no of representations with ext 'abc'"
+
+        assert 2 == dbcon.count_documents({"type": "representation",
+                                           "context.subset": "modelMain",
+                                           "context.ext": "ma"}), \
+            "Not expected no of representations with ext 'abc'"
+
+
+if __name__ == "__main__":
+    test_case = TestDeadlinePublishInMaya()

tests/integration/hosts/maya/test_publish_in_maya.py

@@ -1,7 +1,7 @@
-from tests.integration.hosts.maya.lib import MayaTestClass
+from tests.integration.hosts.maya.lib import MayaLocalPublishTestClass
 
 
-class TestPublishInMaya(MayaTestClass):
+class TestPublishInMaya(MayaLocalPublishTestClass):
     """Basic test case for publishing in Maya
 
         Shouldnt be running standalone only via 'runtests' pype command! (??)

tests/lib/testing_classes.py

@@ -8,9 +8,11 @@ import tempfile
 import shutil
 import glob
 import platform
+import requests
 
 from tests.lib.db_handler import DBHandler
 from common.openpype_common.distribution.file_handler import RemoteFileHandler
+from openpype.modules import ModulesManager
 
 
 class BaseTest:
@@ -333,7 +335,57 @@ class PublishTest(ModuleUnitTest):
             "\n".join(sorted(not_matched)))
 
 
-class HostFixtures(PublishTest):
+class DeadlinePublishTest(PublishTest):
+    @pytest.fixture(scope="module")
+    def publish_finished(self, dbcon, launched_app, download_test_data,
+                         timeout):
+        """Dummy fixture waiting for publish to finish"""
+        import time
+        time_start = time.time()
+        timeout = timeout or self.TIMEOUT
+        timeout = float(timeout)
+        while launched_app.poll() is None:
+            time.sleep(0.5)
+            if time.time() - time_start > timeout:
+                launched_app.terminate()
+                raise ValueError("Timeout reached")
+
+        deadline_job_id = os.environ.get("DEADLINE_PUBLISH_JOB_ID")
+        if not deadline_job_id:
+            raise ValueError("DEADLINE_PUBLISH_JOB_ID empty, cannot find job")
+
+        modules_manager = ModulesManager()
+        deadline_module = modules_manager.modules_by_name("deadline")
+        deadline_url = deadline_module.deadline_urls["default"]
+
+        if not deadline_url:
+            raise ValueError("Must have default deadline url.")
+
+        url = "{}/api/jobs?JobId={}".format(deadline_url, deadline_job_id)
+        date_finished = None
+
+        time_start = time.time()
+        while not date_finished:
+            time.sleep(0.5)
+            if time.time() - time_start > timeout:
+                raise ValueError("Timeout for DL finish reached")
+
+            response = requests.get(url, timeout=10)
+            if not response.ok:
+                msg = "Couldn't connect to {}".format(deadline_url)
+                raise RuntimeError(msg)
+
+            if not response.json():
+                raise ValueError("Couldn't find {}".format(deadline_job_id))
+
+            date_finished = response.json()[0]["DateComp"]
+
+        # some clean exit test possible?
+        print("Publish finished")
+        yield True
+
+
+class HostFixtures():
     """Host specific fixtures. Should be implemented once per host."""
     @pytest.fixture(scope="module")
     def last_workfile_path(self, download_test_data, output_folder_url):