Replace explicit null with default value

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 @@ OutputsDict = Dict[str, OutputChecks]
 class TestJobDict(TypedDict):
     doc: NotRequired[str]
     job: NotRequired[JobDict]
+    expect_failure: NotRequired[bool]
     outputs: OutputsDict
 
 

lib/galaxy/workflow/run.py

@@ -588,7 +588,8 @@ class WorkflowProgress:
                         )
                     )
             elif step_id in self.inputs_by_step_id:
-                outputs["output"] = self.inputs_by_step_id[step_id]
+                if self.inputs_by_step_id[step_id] is not None or "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:
             self.runtime_replacements[step.label] = str(outputs["output"])

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

lib/galaxy_test/workflow/test_framework_workflows.py

@@ -64,14 +64,23 @@ class TestWorkflow(ApiTestCase):
         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():