Merge pull request #17234 from mvdbeek/single_api_instance

Reuse test instance during non-integration tests

lib/galaxy/util/__init__.py

@@ -1734,7 +1734,7 @@ def build_url(base_url, port=80, scheme="http", pathspec=None, params=None, dose
     parsed_url = urlparse(base_url)
     if scheme != "http":
         parsed_url.scheme = scheme
-    assert parsed_url.scheme in ("http", "https", "ftp"), f"Invalid URL scheme: {scheme}"
+    assert parsed_url.scheme in ("http", "https", "ftp"), f"Invalid URL scheme: {parsed_url.scheme}"
     if port != 80:
         url = "%s://%s:%d/%s" % (parsed_url.scheme, parsed_url.netloc.rstrip("/"), int(port), parsed_url.path)
     else:

lib/galaxy_test/api/test_item_tags.py

@@ -8,10 +8,10 @@
     DatasetPopulator,
     WorkflowPopulator,
 )
-from galaxy_test.driver import integration_util
+from ._framework import ApiTestCase
 
 
-class TestItemTagsApi(integration_util.IntegrationTestCase):
+class TestItemTagsApi(ApiTestCase):
     dataset_populator: DatasetPopulator
 
     @classmethod

lib/galaxy_test/base/testcase.py

@@ -1,10 +1,11 @@
 import logging
-import os
 from typing import (
     Any,
     Optional,
 )
 
+import pytest
+
 from galaxy.tool_util.verify.test_data import TestDataResolver
 from galaxy.util.unittest import TestCase
 from galaxy_test.base.env import (
@@ -15,6 +16,7 @@
 log = logging.getLogger(__name__)
 
 
+@pytest.mark.usefixtures("embedded_driver")
 class FunctionalTestCase(TestCase):
     """Base class for tests targetting actual Galaxy servers.
 
@@ -46,17 +48,12 @@ def setUp(self) -> None:
     @classmethod
     def setUpClass(cls):
         """Configure and start Galaxy for a test."""
-        cls._test_driver = None
-
-        if cls.galaxy_driver_class is not None and not os.environ.get("GALAXY_TEST_ENVIRONMENT_CONFIGURED"):
-            cls._test_driver = cls.galaxy_driver_class()
-            cls._test_driver.setup(config_object=cls)
+        pass
 
     @classmethod
     def tearDownClass(cls):
         """Shutdown Galaxy server and cleanup temp directory."""
-        if cls._test_driver:
-            cls._test_driver.tear_down()
+        pass
 
     def get_filename(self, filename: str) -> str:
         # No longer used by tool tests - drop if isn't used else where.

lib/galaxy_test/conftest.py

@@ -8,13 +8,33 @@
 
 from galaxy.util import DEFAULT_SOCKET_TIMEOUT
 from galaxy.web import statsd_client as statsd
+from galaxy_test.api._framework import ApiTestCase
+from galaxy_test.driver.driver_util import GalaxyTestDriver
 
 
 @pytest.fixture(scope="session", autouse=True)
 def celery_includes():
     yield ["galaxy.celery.tasks"]
 
 
+@pytest.fixture(scope="session")
+def real_driver():
+    if not os.environ.get("GALAXY_TEST_ENVIRONMENT_CONFIGURED"):
+        driver = GalaxyTestDriver()
+        driver.setup(ApiTestCase)
+        try:
+            yield driver
+        finally:
+            driver.tear_down()
+    else:
+        yield None
+
+
+@pytest.fixture(scope="class")
+def embedded_driver(real_driver, request):
+    request.cls._test_driver = real_driver
+
+
 def get_timings(test_uuid):
     # timestamps didn't work - telegraf didn't sample at small enough granularity
     # fs = datetime.datetime.fromtimestamp(from_timestamp).strftime('%Y-%m-%dT%H:%M:%S.%fZ')

lib/galaxy_test/selenium/conftest.py

@@ -0,0 +1,24 @@
+import os
+
+import pytest
+
+from galaxy_test.driver.driver_util import GalaxyTestDriver
+from galaxy_test.selenium.framework import SeleniumTestCase
+
+
+@pytest.fixture(scope="session")
+def real_driver():
+    if not os.environ.get("GALAXY_TEST_ENVIRONMENT_CONFIGURED"):
+        driver = GalaxyTestDriver()
+        driver.setup(SeleniumTestCase)
+        try:
+            yield driver
+        finally:
+            driver.tear_down()
+    else:
+        yield None
+
+
+@pytest.fixture(scope="class")
+def embedded_driver(real_driver, request):
+    request.cls._test_driver = real_driver

lib/tool_shed/test/base/api.py

@@ -90,8 +90,8 @@ def tearDownClass(cls):
         pass
 
     @pytest.fixture(autouse=True)
-    def _get_driver(self, tool_shed_test_driver):
-        self._test_driver = tool_shed_test_driver
+    def _get_driver(self, embedded_driver):
+        self._test_driver = embedded_driver
 
 
 class ShedGalaxyInteractorApi(GalaxyInteractorApi):

lib/tool_shed/test/conftest.py

@@ -6,7 +6,7 @@
 
 
 @pytest.fixture(scope="session")
-def tool_shed_test_driver():
+def embedded_driver():
     if not os.environ.get("GALAXY_TEST_ENVIRONMENT_CONFIGURED"):
         ts_driver = driver.ToolShedTestDriver()
         ts_driver.setup()

test/functional/conftest.py

@@ -1,9 +1,29 @@
-def pytest_unconfigure(config):
-    try:
-        # This needs to be run if no test were run.
-        from .test_toolbox_pytest import DRIVER
-
-        DRIVER.tear_down()
-        print("Galaxy test driver shutdown successful")
-    except Exception:
-        pass
+import os
+
+import pytest
+
+from galaxy_test.driver.driver_util import GalaxyTestDriver
+from .test_toolbox_pytest import TestFrameworkTools
+
+
+@pytest.fixture(scope="session", autouse=True)
+def celery_includes():
+    yield ["galaxy.celery.tasks"]
+
+
+@pytest.fixture(scope="session")
+def real_driver():
+    if not os.environ.get("GALAXY_TEST_ENVIRONMENT_CONFIGURED"):
+        driver = GalaxyTestDriver()
+        driver.setup(TestFrameworkTools)
+        try:
+            yield driver
+        finally:
+            driver.tear_down()
+    else:
+        yield None
+
+
+@pytest.fixture(scope="class")
+def embedded_driver(real_driver, request):
+    request.cls._test_driver = real_driver

test/functional/test_toolbox_pytest.py

@@ -1,16 +1,13 @@
 import os
-import sys
 from typing import (
     List,
     NamedTuple,
 )
 
-galaxy_root = os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir + "/" + os.path.pardir))
-sys.path[1:1] = [os.path.join(galaxy_root, "lib"), os.path.join(galaxy_root, "test")]
-
 import pytest
 
 from galaxy_test.api._framework import ApiTestCase
+from galaxy_test.driver.driver_util import GalaxyTestDriver
 
 SKIPTEST = os.path.join(os.path.dirname(__file__), "known_broken_tools.txt")
 
@@ -29,16 +26,16 @@ def get_skiplist():
 
 def get_cases() -> List[ToolTest]:
     atc = ApiTestCase()
-    atc.setUpClass()
+    atc._test_driver = GalaxyTestDriver()
+    atc._test_driver.setup()
     atc.setUp()
     test_summary = atc.galaxy_interactor.get_tests_summary()
     test_cases = []
     for tool_id, summary_dict in test_summary.items():
         for tool_version, tool_dict in summary_dict.items():
             for index in range(tool_dict["count"]):
                 test_cases.append(ToolTest(tool_id, tool_version, index))
-    atc.tearDown()
-    atc.tearDownClass()
+    atc._test_driver.stop_servers()
     return test_cases