Reduce the number of test workflow imports

Author: nsoranzo

bioblend/_tests/GalaxyTestBase.py

@@ -20,7 +20,7 @@ class GalaxyTestBase(unittest.TestCase):
     gi: GalaxyInstance
 
     @classmethod
-    def setUpClass(cls):
+    def setUpClass(cls) -> None:
         galaxy_key = os.environ["BIOBLEND_GALAXY_API_KEY"]
         galaxy_url = os.environ["BIOBLEND_GALAXY_URL"]
         cls.gi = GalaxyInstance(url=galaxy_url, key=galaxy_key)

bioblend/_tests/TestGalaxyInvocations.py

@@ -13,14 +13,21 @@
 
 
 class TestGalaxyInvocations(GalaxyTestBase.GalaxyTestBase):
+    workflow_id: str
+    pause_workflow_id: str
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        super().setUpClass()
+        path = test_util.get_abspath(os.path.join("data", "paste_columns.ga"))
+        cls.workflow_id = cls.gi.workflows.import_workflow_from_local_path(path)["id"]
+        path = test_util.get_abspath(os.path.join("data", "test_workflow_pause.ga"))
+        cls.pause_workflow_id = cls.gi.workflows.import_workflow_from_local_path(path)["id"]
+
     def setUp(self):
         super().setUp()
-        path = test_util.get_abspath(os.path.join("data", "paste_columns.ga"))
-        self.workflow_id = self.gi.workflows.import_workflow_from_local_path(path)["id"]
         self.history_id = self.gi.histories.create_history(name="TestGalaxyInvocations")["id"]
         self.dataset_id = self._test_dataset(self.history_id)
-        path = test_util.get_abspath(os.path.join("data", "test_workflow_pause.ga"))
-        self.pause_workflow_id = self.gi.workflows.import_workflow_from_local_path(path)["id"]
 
     def tearDown(self):
         self.gi.histories.delete_history(self.history_id, purge=True)
@@ -42,16 +49,16 @@ def test_cancel_invocation(self):
     def test_get_invocations(self):
         invoc1 = self._invoke_workflow()
 
-        # Run the first workflow on another history
+        # Run another workflow on the same history
+        path = test_util.get_abspath(os.path.join("data", "paste_columns.ga"))
+        workflow2_id = self.gi.workflows.import_workflow_from_local_path(path)["id"]
         dataset = {"src": "hda", "id": self.dataset_id}
-        hist2_id = self.gi.histories.create_history("hist2")["id"]
         invoc2 = self.gi.workflows.invoke_workflow(
-            self.workflow_id, history_id=hist2_id, inputs={"Input 1": dataset, "Input 2": dataset}, inputs_by="name"
+            workflow2_id, history_id=self.history_id, inputs={"Input 1": dataset, "Input 2": dataset}, inputs_by="name"
         )
 
-        # Run another workflow on the 2nd history
-        path = test_util.get_abspath(os.path.join("data", "paste_columns.ga"))
-        workflow2_id = self.gi.workflows.import_workflow_from_local_path(path)["id"]
+        # Run the second workflow on another history
+        hist2_id = self.gi.histories.create_history("hist2")["id"]
         invoc3 = self.gi.workflows.invoke_workflow(
             workflow2_id, history_id=hist2_id, inputs={"Input 1": dataset, "Input 2": dataset}, inputs_by="name"
         )
@@ -60,14 +67,13 @@ def test_get_invocations(self):
             self.gi.invocations.wait_for_invocation(invoc["id"])
 
         # Test filtering by workflow ID
-        for wf_id, expected_invoc_num in {self.workflow_id: 2, workflow2_id: 1}.items():
-            invocs = self.gi.invocations.get_invocations(workflow_id=wf_id)
-            assert len(invocs) == expected_invoc_num
-            for invoc in invocs:
-                assert invoc["workflow_id"] == wf_id
+        invocs = self.gi.invocations.get_invocations(workflow_id=workflow2_id)
+        assert len(invocs) == 2
+        for invoc in invocs:
+            assert invoc["workflow_id"] == workflow2_id
 
         # Test filtering by history ID
-        for hist_id, expected_invoc_num in {self.history_id: 1, hist2_id: 2}.items():
+        for hist_id, expected_invoc_num in {self.history_id: 2, hist2_id: 1}.items():
             invocs = self.gi.invocations.get_invocations(history_id=hist_id)
             assert len(invocs) == expected_invoc_num
             for invoc in invocs:

bioblend/_tests/TestGalaxyWorkflows.py

@@ -93,8 +93,7 @@ def test_invoke_workflow_parameters_normalized(self):
     @test_util.skip_unless_tool("cat")
     def test_cancelling_workflow_scheduling(self):
         path = test_util.get_abspath(os.path.join("data", "test_workflow_pause.ga"))
-        workflow = self.gi.workflows.import_workflow_from_local_path(path)
-        workflow_id = workflow["id"]
+        workflow_id = self.gi.workflows.import_workflow_from_local_path(path)["id"]
         history_id = self.gi.histories.create_history(name="TestWorkflowState")["id"]
         dataset1_id = self._test_dataset(history_id)