From e4fc02049e879929e89c441cdac0d245126d16f3 Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Wed, 6 Nov 2024 11:07:40 +0000 Subject: [PATCH] Reduce the number of test workflow imports --- bioblend/_tests/GalaxyTestBase.py | 2 +- bioblend/_tests/TestGalaxyInvocations.py | 38 ++++++++++++++---------- bioblend/_tests/TestGalaxyWorkflows.py | 3 +- 3 files changed, 24 insertions(+), 19 deletions(-) diff --git a/bioblend/_tests/GalaxyTestBase.py b/bioblend/_tests/GalaxyTestBase.py index 351cf6d91..7e1c473b8 100644 --- a/bioblend/_tests/GalaxyTestBase.py +++ b/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) diff --git a/bioblend/_tests/TestGalaxyInvocations.py b/bioblend/_tests/TestGalaxyInvocations.py index 0748ae9c7..d9bc02db5 100644 --- a/bioblend/_tests/TestGalaxyInvocations.py +++ b/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: diff --git a/bioblend/_tests/TestGalaxyWorkflows.py b/bioblend/_tests/TestGalaxyWorkflows.py index 75575dc18..000506efc 100644 --- a/bioblend/_tests/TestGalaxyWorkflows.py +++ b/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)