Rework uses_shed...

lib/galaxy_test/base/uses_shed_api.py

@@ -0,0 +1,42 @@
+from unittest import SkipTest
+
+from galaxy.tool_util.verify.interactor import GalaxyInteractorApi
+from galaxy.util import unicodify
+from galaxy_test.base.api_asserts import assert_status_code_is
+
+
+class UsesShedApi:
+    galaxy_interactor: GalaxyInteractorApi
+
+    def delete_repo_request(self, payload):
+        return self.galaxy_interactor._delete("tool_shed_repositories", data=payload, admin=True)
+
+    def install_repo_request(self, payload):
+        return self.galaxy_interactor._post(
+            "tool_shed_repositories/new/install_repository_revision", data=payload, admin=True
+        )
+
+    def repository_operation(self, operation, owner, name, changeset, tool_shed_url="https://toolshed.g2.bx.psu.edu"):
+        payload = {"tool_shed_url": tool_shed_url, "name": name, "owner": owner, "changeset_revision": changeset}
+        create_response = operation(payload)
+        assert_status_code_is(create_response, 200)
+        return create_response.json()
+
+    def install_repository(self, owner, name, changeset, tool_shed_url="https://toolshed.g2.bx.psu.edu"):
+        try:
+            return self.repository_operation(
+                operation=self.install_repo_request,
+                owner=owner,
+                name=name,
+                changeset=changeset,
+                tool_shed_url=tool_shed_url,
+            )
+        except AssertionError as e:
+            if "Error attempting to retrieve installation information from tool shed" in unicodify(e):
+                raise SkipTest(f"Toolshed '{tool_shed_url}' unavailable")
+            raise
+
+    def uninstall_repository(self, owner, name, changeset, tool_shed_url="https://toolshed.g2.bx.psu.edu"):
+        return self.repository_operation(
+            operation=self.delete_repo_request, owner=owner, name=name, changeset=changeset, tool_shed_url=tool_shed_url
+        )

lib/galaxy_test/driver/uses_shed.py

@@ -4,13 +4,9 @@ import string
 import tempfile
 from typing import ClassVar
 
-import pytest
-
 from galaxy.app import UniverseApplication
-from galaxy.tool_util.verify.interactor import GalaxyInteractorApi
-from galaxy.util import unicodify
-from galaxy_test.base.api_asserts import assert_status_code_is
 from galaxy_test.base.populators import DEFAULT_TIMEOUT
+from galaxy_test.base.uses_shed_api import UsesShedApi
 from galaxy_test.driver.driver_util import (
     FRAMEWORK_UPLOAD_TOOL_CONF,
     GalaxyTestDriver,
@@ -37,7 +33,7 @@ SHED_DATA_TABLES = """<?xml version="1.0"?>
 </tables>"""
 
 
-class UsesShed:
+class UsesShed(UsesShedApi):
     @property
     def _app(self) -> UniverseApplication:
         ...
@@ -45,7 +41,6 @@ class UsesShed:
     shed_tools_dir: ClassVar[str]
     shed_tool_data_dir: ClassVar[str]
     conda_tmp_prefix: ClassVar[str]
-    galaxy_interactor: GalaxyInteractorApi
     _test_driver: GalaxyTestDriver
 
     @classmethod
@@ -101,36 +96,3 @@ class UsesShed:
         for item in models_to_delete:
             model.context.query(item).delete()
         model.context.flush()
-
-    def delete_repo_request(self, payload):
-        return self.galaxy_interactor._delete("tool_shed_repositories", data=payload, admin=True)
-
-    def install_repo_request(self, payload):
-        return self.galaxy_interactor._post(
-            "tool_shed_repositories/new/install_repository_revision", data=payload, admin=True
-        )
-
-    def repository_operation(self, operation, owner, name, changeset, tool_shed_url="https://toolshed.g2.bx.psu.edu"):
-        payload = {"tool_shed_url": tool_shed_url, "name": name, "owner": owner, "changeset_revision": changeset}
-        create_response = operation(payload)
-        assert_status_code_is(create_response, 200)
-        return create_response.json()
-
-    def install_repository(self, owner, name, changeset, tool_shed_url="https://toolshed.g2.bx.psu.edu"):
-        try:
-            return self.repository_operation(
-                operation=self.install_repo_request,
-                owner=owner,
-                name=name,
-                changeset=changeset,
-                tool_shed_url=tool_shed_url,
-            )
-        except AssertionError as e:
-            if "Error attempting to retrieve installation information from tool shed" in unicodify(e):
-                pytest.skip(f"Toolshed '{tool_shed_url}' unavailable")
-            raise
-
-    def uninstall_repository(self, owner, name, changeset, tool_shed_url="https://toolshed.g2.bx.psu.edu"):
-        return self.repository_operation(
-            operation=self.delete_repo_request, owner=owner, name=name, changeset=changeset, tool_shed_url=tool_shed_url
-        )

test/integration/test_workflow_invocation.py

@@ -4,11 +4,11 @@ from galaxy_test.base.populators import (
     DatasetPopulator,
     WorkflowPopulator,
 )
+from galaxy_test.base.uses_shed_api import UsesShedApi
 from galaxy_test.driver import integration_util
-from galaxy_test.driver.uses_shed import UsesShed
 
 
-class WorkflowInvocationTestCase(integration_util.IntegrationTestCase, UsesShed):
+class WorkflowInvocationTestCase(integration_util.IntegrationTestCase, UsesShedApi):
 
     framework_tool_and_types = True
     require_admin_user = False

test/integration/test_workflow_refactoring.py

@@ -15,13 +15,13 @@ from galaxy.model import (
 )
 from galaxy.workflow.refactor.schema import RefactorActionExecutionMessageTypeEnum
 from galaxy_test.base.populators import WorkflowPopulator
+from galaxy_test.base.uses_shed_api import UsesShedApi
 from galaxy_test.base.workflow_fixtures import (
     WORKFLOW_NESTED_RUNTIME_PARAMETER,
     WORKFLOW_NESTED_SIMPLE,
     WORKFLOW_NESTED_WITH_MULTIPLE_VERSIONS_TOOL,
 )
 from galaxy_test.driver import integration_util
-from galaxy_test.driver.uses_shed import UsesShed
 
 REFACTORING_SIMPLE_TEST = """
 class: GalaxyWorkflow
@@ -38,7 +38,7 @@ steps:
 """
 
 
-class WorkflowRefactoringIntegrationTestCase(integration_util.IntegrationTestCase, UsesShed):
+class WorkflowRefactoringIntegrationTestCase(integration_util.IntegrationTestCase, UsesShedApi):
 
     framework_tool_and_types = True