Replace explicit null with default value

galaxyproject · Nov 29, 2024 · 7ab0178 · 7ab0178
1 parent 7867388
commit 7ab0178
Show file tree

Hide file tree

Showing 4 changed files with 22 additions and 10 deletions.
diff --git a/lib/galaxy/tool_util/models.py b/lib/galaxy/tool_util/models.py
@@ -172,6 +172,7 @@ class TestJob(StrictModel):
     doc: Optional[str]
     job: JobDict
     outputs: Dict[str, TestOutputAssertions]
+    expect_failure: Optional[bool] = False
 
 
 Tests = RootModel[List[TestJob]]
@@ -185,6 +186,7 @@ class TestJob(StrictModel):
 class TestJobDict(TypedDict):
     doc: NotRequired[str]
     job: NotRequired[JobDict]
+    expect_failure: NotRequired[bool]
     outputs: OutputsDict
 
 

diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py
@@ -587,7 +587,7 @@ def set_outputs_for_input(
                             dependent_workflow_step_id=invocation_step.workflow_step_id,
                         )
                     )
-            elif step_id in self.inputs_by_step_id:
+            elif step_id in self.inputs_by_step_id and "output" not in outputs:
                 outputs["output"] = self.inputs_by_step_id[step_id]
 
         if step.label and step.type == "parameter_input" and "output" in outputs:

diff --git a/lib/galaxy_test/workflow/default_values.gxwf-tests.yml b/lib/galaxy_test/workflow/default_values.gxwf-tests.yml
@@ -8,7 +8,7 @@
       - that: has_text
         text: "1"
 - doc: |
-    Test that null is replaced with default value
+    Test that null is replaced with default value (follows https://www.commonwl.org/v1.2/Workflow.html#WorkflowInputParameter)
   job:
     required_int_with_default:
       type: raw
@@ -21,6 +21,7 @@
         text: "1"
 - doc: |
     Test that empty string is not replaced and fails
+  expect_failure: true
   job:
     required_int_with_default:
       type: raw

diff --git a/lib/galaxy_test/workflow/test_framework_workflows.py b/lib/galaxy_test/workflow/test_framework_workflows.py
@@ -64,14 +64,23 @@ def test_workflow(self, workflow_path: Path, test_job: TestJobDict):
         with workflow_path.open() as f:
             yaml_content = ordered_load(f)
         with self.dataset_populator.test_history() as history_id:
-            run_summary = self.workflow_populator.run_workflow(
-                yaml_content,
-                test_data=test_job["job"],
-                history_id=history_id,
-            )
-            if TEST_WORKFLOW_AFTER_RERUN:
-                run_summary = self.workflow_populator.rerun(run_summary)
-            self._verify(run_summary, test_job["outputs"])
+            exc = None
+            try:
+                run_summary = self.workflow_populator.run_workflow(
+                    yaml_content,
+                    test_data=test_job["job"],
+                    history_id=history_id,
+                )
+                if TEST_WORKFLOW_AFTER_RERUN:
+                    run_summary = self.workflow_populator.rerun(run_summary)
+                self._verify(run_summary, test_job["outputs"])
+            except Exception as e:
+                exc = e
+            if test_job.get("expect_failure"):
+                if not exc:
+                    raise Exception("Expected workflow test to fail but it passed")
+            elif exc:
+                raise exc
 
     def _verify(self, run_summary: RunJobsSummary, output_definitions: OutputsDict):
         for output_name, output_definition in output_definitions.items():