Add integration test for job failure behavior when a tool is removed while a job is queued

2026-09-24 16:30:27 +08:00 · 2019-06-18 17:47:09 +02:00
parent 4f22789b01
commit d0bf7c9767
1 changed files with 55 additions and 0 deletions
@@ -0,0 +1,55 @@
+import time
+from base import integration_util
+from base.populators import (
+    DatasetPopulator,
+    WorkflowPopulator,
+)
+
+
+class FailJobWhenToolUnavailableTestCase(integration_util.IntegrationTestCase):
+
+    require_admin_user = True
+
+    def setUp(self):
+        super(FailJobWhenToolUnavailableTestCase, self).setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+        self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
+        self.history_id = self.dataset_populator.new_history()
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config, ):
+        # config["jobs_directory"] = cls.jobs_directory
+        # Disable tool dependency resolution.
+        config["tool_dependency_dir"] = "none"
+
+    def test_fail_job_when_tool_unavailable(self):
+        self.workflow_populator.run_workflow("""
+class: GalaxyWorkflow
+steps:
+  - label: sleep
+    run:
+      class: GalaxyTool
+      command: sleep 20s && echo 'hello world 2' > '$output1'
+      outputs:
+        output1:
+          format: txt
+  - tool_id: cat1
+    state:
+      input1:
+        $link: sleep#output1
+      queries:
+        input2:
+          $link: sleep#output1
+""", history_id=self.history_id, assert_ok=False, wait=False)
+        # Wait until workflow is fully scheduled, otherwise can't test effect of removing tool from queued job
+        time.sleep(10)
+        self._app.toolbox.remove_tool_by_id('cat1')
+        self.dataset_populator.wait_for_history(self.history_id, assert_ok=False)
+        state_details = self.galaxy_interactor.get('histories/%s' % self.history_id).json()['state_details']
+        assert state_details['running'] == 0
+        assert state_details['ok'] == 1
+        assert state_details['error'] == 1
+        failed_hda = self.dataset_populator.get_history_dataset_details(history_id=self.history_id, assert_ok=False, details=True)
+        assert failed_hda['state'] == 'error'
+        job = self.galaxy_interactor.get("jobs/%s" % failed_hda['creating_job']).json()
+        assert job['state'] == 'error'