Type annotation and refactoring of tests

lib/galaxy_test/driver/driver_util.py

@@ -15,6 +15,8 @@
 import time
 from pathlib import Path
 from typing import (
+    Any,
+    Dict,
     List,
     Optional,
 )
@@ -32,7 +34,6 @@
     GalaxyInteractorApi,
     verify_tool,
 )
-from galaxy.tools import ToolBox
 from galaxy.util import (
     asbool,
     download_to_file,
@@ -538,7 +539,7 @@ def run_in_loop(loop):
     return server, port, t
 
 
-def cleanup_directory(tempdir):
+def cleanup_directory(tempdir: str) -> None:
     """Clean up temporary files used by test unless GALAXY_TEST_NO_CLEANUP is set.
 
     Also respect TOOL_SHED_TEST_NO_CLEANUP for legacy reasons.
@@ -554,24 +555,6 @@ def cleanup_directory(tempdir):
         pass
 
 
-def setup_shed_tools_for_test(app, tmpdir, testing_migrated_tools, testing_installed_tools):
-    """Modify Galaxy app's toolbox for migrated or installed tool tests."""
-    if testing_installed_tools:
-        # TODO: Do this without modifying app - that is a pretty violation
-        # of Galaxy's abstraction - we shouldn't require app at all let alone
-        # be modifying it.
-
-        tool_configs = app.config.tool_configs
-        # Eliminate the migrated_tool_panel_config from the app's tool_configs, append the list of installed_tool_panel_configs,
-        # and reload the app's toolbox.
-        relative_migrated_tool_panel_config = os.path.join(app.config.root, MIGRATED_TOOL_PANEL_CONFIG)
-        if relative_migrated_tool_panel_config in tool_configs:
-            tool_configs.remove(relative_migrated_tool_panel_config)
-        for installed_tool_panel_config in INSTALLED_TOOL_PANEL_CONFIGS:
-            tool_configs.append(installed_tool_panel_config)
-        app.toolbox = ToolBox(tool_configs, app.config.tool_path, app)
-
-
 def build_galaxy_app(simple_kwargs) -> GalaxyUniverseApplication:
     """Build a Galaxy app object from a simple keyword arguments.
 
@@ -667,7 +650,7 @@ def get_logs(self) -> Optional[str]:
     def app(self):
         raise NotImplementedError("Test can be run against target - requires a Galaxy app object.")
 
-    def stop(self):
+    def stop(self) -> None:
         raise NotImplementedError()
 
 
@@ -834,22 +817,19 @@ class TestDriver:
 
     def __init__(self):
         """Setup tracked resources."""
-        self.server_wrappers = []
-        self.temp_directories = []
+        self.server_wrappers: List[ServerWrapper] = []
+        self.temp_directories: List[str] = []
 
-    def setup(self, config_object=None):
+    def setup(self, config_object=None) -> None:
         """Called before tests are built."""
 
-    def build_tests(self):
-        """After environment is setup, setup tests."""
-
-    def tear_down(self):
+    def tear_down(self) -> None:
         """Cleanup resources tracked by this object."""
         self.stop_servers()
         for temp_directory in self.temp_directories:
             cleanup_directory(temp_directory)
 
-    def stop_servers(self):
+    def stop_servers(self) -> None:
         for server_wrapper in self.server_wrappers:
             server_wrapper.stop()
         for th in threading.enumerate():
@@ -872,10 +852,9 @@ def mkdtemp(self) -> str:
 class GalaxyTestDriver(TestDriver):
     """Instantial a Galaxy-style TestDriver for testing Galaxy."""
 
-    server_wrappers: List[ServerWrapper]
     testing_shed_tools = False
 
-    def _configure(self, config_object=None):
+    def _configure(self, config_object=None) -> None:
         """Setup various variables used to launch a Galaxy server."""
         config_object = self._ensure_config_object(config_object)
         self.external_galaxy = os.environ.get("GALAXY_TEST_EXTERNAL", None)
@@ -913,18 +892,19 @@ def setup(self, config_object=None):
         self._configure(config_object)
         self._register_and_run_servers(config_object)
 
-    def get_logs(self):
-        if self.server_wrappers:
-            server_wrapper = self.server_wrappers[0]
-            return server_wrapper.get_logs()
+    def get_logs(self) -> Optional[str]:
+        if not self.server_wrappers:
+            return None
+        server_wrapper = self.server_wrappers[0]
+        return server_wrapper.get_logs()
 
-    def restart(self, config_object=None, handle_config=None):
+    def restart(self, config_object=None, handle_config=None) -> None:
         self.stop_servers()
         self._register_and_run_servers(config_object, handle_config=handle_config)
 
-    def _register_and_run_servers(self, config_object=None, handle_config=None):
+    def _register_and_run_servers(self, config_object=None, handle_config=None) -> None:
         config_object = self._ensure_config_object(config_object)
-        self.app = None
+        self.app: Optional[GalaxyUniverseApplication] = None
 
         if self.external_galaxy is None:
             if self._saved_galaxy_config is not None:
@@ -978,7 +958,7 @@ def _register_and_run_servers(self, config_object=None, handle_config=None):
             # Ensure test file directory setup even though galaxy config isn't built.
             ensure_test_file_dir_set()
 
-    def build_galaxy_app(self, galaxy_config):
+    def build_galaxy_app(self, galaxy_config) -> GalaxyUniverseApplication:
         self.app = build_galaxy_app(galaxy_config)
         return self.app
 
@@ -987,38 +967,12 @@ def _ensure_config_object(self, config_object):
             config_object = self
         return config_object
 
-    def setup_shed_tools(self, testing_migrated_tools=False, testing_installed_tools=True):
-        setup_shed_tools_for_test(self.app, self.galaxy_test_tmp_dir, testing_migrated_tools, testing_installed_tools)
-
-    def build_tool_tests(self, testing_shed_tools=None, return_test_classes=False):
-        if self.app is None:
-            return
-
-        if testing_shed_tools is None:
-            testing_shed_tools = getattr(self, "testing_shed_tools", False)
-
-        # We must make sure that functional.test_toolbox is always imported after
-        # database_contexts.galaxy_content is set (which occurs in this method above).
-        # If functional.test_toolbox is imported before database_contexts.galaxy_content
-        # is set, sa_session will be None in all methods that use it.
-        import functional.test_toolbox
-
-        functional.test_toolbox.toolbox = self.app.toolbox
-        # When testing data managers, do not test toolbox.
-        test_classes = functional.test_toolbox.build_tests(
-            app=self.app,
-            testing_shed_tools=testing_shed_tools,
-            master_api_key=get_admin_api_key(),
-            user_api_key=get_user_api_key(),
-        )
-        if return_test_classes:
-            return test_classes
-        return functional.test_toolbox
-
-    def run_tool_test(self, tool_id, index=0, resource_parameters=None, **kwd):
+    def run_tool_test(
+        self, tool_id: str, index: int = 0, resource_parameters: Optional[Dict[str, Any]] = None, **kwd
+    ) -> None:
         if resource_parameters is None:
             resource_parameters = {}
-        host, port, url = target_url_parts()
+        _, _, url = target_url_parts()
         galaxy_interactor_kwds = {
             "galaxy_url": url,
             "master_api_key": get_admin_api_key(),

lib/galaxy_test/driver/integration_util.py

@@ -13,7 +13,6 @@
     Optional,
     Type,
     TYPE_CHECKING,
-    TypeVar,
 )
 from unittest import (
     skip,
@@ -198,16 +197,16 @@ class IntegrationTestCase(IntegrationInstance, TestCase):
     """Unit TestCase with utilities for spinning up Galaxy."""
 
 
-IntegrationInstanceObject = TypeVar("IntegrationInstanceObject", bound=IntegrationInstance)
-
-
-def integration_module_instance(clazz: Type[IntegrationInstanceObject]):
-    def _instance() -> Iterator[IntegrationInstanceObject]:
+def integration_module_instance(clazz: Type[IntegrationInstance]):
+    def _instance() -> Iterator[IntegrationInstance]:
         instance = clazz()
         instance.setUpClass()
         instance.setUp()
-        yield instance
-        instance.tearDownClass()
+        try:
+            yield instance
+        finally:
+            instance.tearDown()
+            instance.tearDownClass()
 
     return pytest.fixture(scope="module")(_instance)
 

test/integration/objectstore/test_objectstore_datatype_upload.py

@@ -8,12 +8,13 @@
 
 import pytest
 
+from galaxy.objectstore.irods import IRODSObjectStore
 from galaxy_test.driver import integration_util
 from ..test_datatype_upload import (
     TEST_CASES,
     TestData,
     upload_datatype_helper,
-    UploadTestDatatypeDataTestCase,
+    UploadTestDatatypeDataIntegrationInstance,
 )
 
 REFRESH_TIME = 3
@@ -118,7 +119,7 @@ def stop_irods(container_name):
         subprocess.check_call(["docker", "rm", "-f", container_name])
 
 
-class BaseObjectstoreUploadTest(UploadTestDatatypeDataTestCase):
+class BaseObjectstoreUploadIntegrationInstance(UploadTestDatatypeDataIntegrationInstance):
     object_store_template: Optional[string.Template] = None
 
     @classmethod
@@ -139,19 +140,19 @@ def get_object_store_kwargs(cls):
         return {}
 
 
-class IrodsUploadTestDatatypeDataTestCase(BaseObjectstoreUploadTest):
+class IrodsUploadTestDatatypeDataIntegrationInstance(BaseObjectstoreUploadIntegrationInstance):
     object_store_template = IRODS_OBJECT_STORE_CONFIG
 
     @classmethod
     def setUpClass(cls):
         cls.container_name = "irods_integration_container"
         start_irods(cls.container_name)
-        super(UploadTestDatatypeDataTestCase, cls).setUpClass()
+        super().setUpClass()
 
     @classmethod
     def tearDownClass(cls):
         stop_irods(cls.container_name)
-        super(UploadTestDatatypeDataTestCase, cls).tearDownClass()
+        super().tearDownClass()
 
     @classmethod
     def get_object_store_kwargs(cls):
@@ -169,61 +170,66 @@ def get_object_store_kwargs(cls):
         }
 
 
-class IrodsIdleConnectionUploadTestCase(IrodsUploadTestDatatypeDataTestCase):
+class IrodsIdleConnectionUploadIntegrationInstance(IrodsUploadTestDatatypeDataIntegrationInstance):
     object_store_template = IRODS_OBJECT_STORE_CONFIG
 
 
-class UploadTestDosDiskAndDiskTestCase(BaseObjectstoreUploadTest):
+class UploadTestDosDiskAndDiskIntegrationInstance(BaseObjectstoreUploadIntegrationInstance):
     object_store_template = DISTRIBUTED_OBJECT_STORE_CONFIG
 
     @classmethod
     def get_object_store_kwargs(cls):
         return {"temp_directory": cls.object_stores_parent}
 
 
-class UploadTestDosIrodsAndDiskTestCase(IrodsUploadTestDatatypeDataTestCase):
+class UploadTestDosIrodsAndDiskIntegrationInstance(IrodsUploadTestDatatypeDataIntegrationInstance):
     object_store_template = DISTRIBUTED_IRODS_OBJECT_STORE_CONFIG
 
 
-distributed_instance = integration_util.integration_module_instance(UploadTestDosDiskAndDiskTestCase)
-irods_instance = integration_util.integration_module_instance(IrodsUploadTestDatatypeDataTestCase)
-distributed_and_irods_instance = integration_util.integration_module_instance(UploadTestDosIrodsAndDiskTestCase)
-idle_connection_irods_instance = integration_util.integration_module_instance(IrodsIdleConnectionUploadTestCase)
+distributed_instance = integration_util.integration_module_instance(UploadTestDosDiskAndDiskIntegrationInstance)
+irods_instance = integration_util.integration_module_instance(IrodsUploadTestDatatypeDataIntegrationInstance)
+distributed_and_irods_instance = integration_util.integration_module_instance(
+    UploadTestDosIrodsAndDiskIntegrationInstance
+)
+idle_connection_irods_instance = integration_util.integration_module_instance(
+    IrodsIdleConnectionUploadIntegrationInstance
+)
 
 
 @pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys()))
 def test_upload_datatype_dos_disk_and_disk(
-    distributed_instance: UploadTestDosDiskAndDiskTestCase, test_data: TestData, temp_file
+    distributed_instance: UploadTestDosDiskAndDiskIntegrationInstance, test_data: TestData, temp_file
 ) -> None:
     with distributed_instance.dataset_populator.test_history() as history_id:
         upload_datatype_helper(distributed_instance, test_data, temp_file, history_id)
 
 
 @pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys()))
 def test_upload_datatype_irods(
-    irods_instance: IrodsUploadTestDatatypeDataTestCase, test_data: TestData, temp_file
+    irods_instance: IrodsUploadTestDatatypeDataIntegrationInstance, test_data: TestData, temp_file
 ) -> None:
     with irods_instance.dataset_populator.test_history() as history_id:
         upload_datatype_helper(irods_instance, test_data, temp_file, history_id, True)
 
 
 @pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys()))
 def test_upload_datatype_dos_irods_and_disk(
-    distributed_and_irods_instance: UploadTestDosIrodsAndDiskTestCase, test_data: TestData, temp_file
+    distributed_and_irods_instance: UploadTestDosIrodsAndDiskIntegrationInstance, test_data: TestData, temp_file
 ) -> None:
     with distributed_and_irods_instance.dataset_populator.test_history() as history_id:
         upload_datatype_helper(distributed_and_irods_instance, test_data, temp_file, history_id)
 
 
 @pytest.mark.parametrize("test_data", SINGLE_TEST_CASE.values(), ids=list(SINGLE_TEST_CASE.keys()))
 def test_upload_datatype_irods_idle_connections(
-    idle_connection_irods_instance: IrodsIdleConnectionUploadTestCase, test_data: TestData, temp_file
+    idle_connection_irods_instance: IrodsIdleConnectionUploadIntegrationInstance, test_data: TestData, temp_file
 ) -> None:
     with idle_connection_irods_instance.dataset_populator.test_history() as history_id:
         upload_datatype_helper(idle_connection_irods_instance, test_data, temp_file, history_id, True)
 
     # Get Irods object store's connection pool
     assert idle_connection_irods_instance._test_driver.app
+    assert isinstance(idle_connection_irods_instance._test_driver.app.object_store, IRODSObjectStore)
     connection_pool = idle_connection_irods_instance._test_driver.app.object_store.session.pool
 
     # Verify the connection pool has 0 active and 1 idle connections

test/integration/test_datatype_upload.py

@@ -14,7 +14,7 @@
 )
 from galaxy.util.hash_util import md5_hash_file
 from galaxy_test.driver import integration_util
-from .test_upload_configuration_options import BaseUploadContentConfigurationInstance
+from .test_upload_configuration_options import BaseUploadContentConfigurationIntegrationInstance
 
 SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
 TEST_FILE_DIR = f"{SCRIPT_DIRECTORY}/../../lib/galaxy/datatypes/test"
@@ -42,14 +42,14 @@ def collect_test_data(registry):
     return {os.path.basename(data.path): data for data in test_data_description}
 
 
-class UploadTestDatatypeDataTestCase(BaseUploadContentConfigurationInstance):
+class UploadTestDatatypeDataIntegrationInstance(BaseUploadContentConfigurationIntegrationInstance):
     framework_tool_and_types = False
     datatypes_conf_override = DATATYPES_CONFIG
     object_store_config = None
     object_store_config_path = None
 
 
-instance = integration_util.integration_module_instance(UploadTestDatatypeDataTestCase)
+instance = integration_util.integration_module_instance(UploadTestDatatypeDataIntegrationInstance)
 
 
 registry = Registry()
@@ -59,7 +59,7 @@ class UploadTestDatatypeDataTestCase(BaseUploadContentConfigurationInstance):
 
 @pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys()))
 def test_upload_datatype_auto(
-    instance: UploadTestDatatypeDataTestCase,
+    instance: UploadTestDatatypeDataIntegrationInstance,
     test_data: TestData,
     temp_file,
     celery_session_worker,
@@ -70,7 +70,7 @@ def test_upload_datatype_auto(
 
 
 def upload_datatype_helper(
-    instance: UploadTestDatatypeDataTestCase,
+    instance: UploadTestDatatypeDataIntegrationInstance,
     test_data: TestData,
     temp_file,
     history_id: str,
@@ -119,9 +119,9 @@ def upload_datatype_helper(
             # Delete cache directory and then re-create it. This way we confirm
             # that dataset is fetched from the object store, not from the cache
             if TYPE_CHECKING:
-                from .objectstore.test_objectstore_datatype_upload import BaseObjectstoreUploadTest
+                from .objectstore.test_objectstore_datatype_upload import BaseObjectstoreUploadIntegrationInstance
 
-                assert isinstance(instance, BaseObjectstoreUploadTest)
+                assert isinstance(instance, BaseObjectstoreUploadIntegrationInstance)
             temp_dir = instance.get_object_store_kwargs()["temp_directory"]
             cache_dir = temp_dir + "/object_store_cache"
             shutil.rmtree(cache_dir)