Re-allow workflow outputs that aren't produced during workflow scheduling.

These are sort of an error in the workflow definition but we don't have a means to competently display this information to the user via the UI so I think we need to just allow this to happen.

I've attached a test case that exhibits this behavior and passes under 17.09 but fails under the current dev branch without these fixes. This was broken in https://github.com/galaxyproject/galaxy/commit/d5f98f49045bca351e982f17cac9386308089eeb when we started to actually persist pointers to workflow invocation outputs (as part of https://github.com/galaxyproject/galaxy/pull/5013).

lib/galaxy/workflow/run.py

@@ -400,7 +400,15 @@ class WorkflowProgress(object):
             for workflow_output in step.workflow_outputs:
                 output_name = workflow_output.output_name
                 if output_name not in outputs:
-                    raise KeyError("Failed to find [%s] in step outputs [%s]" % (output_name, outputs))
+                    message = "Failed to find expected workflow output [%s] in step outputs [%s]" % (output_name, outputs)
+                    # raise KeyError(message)
+                    # Pre-18.01 we would have never even detected this output wasn't configured
+                    # and even in 18.01 we don't have a way to tell the user something bad is
+                    # happening so I guess we just log a debug message and continue sadly for now.
+                    # Once https://github.com/galaxyproject/galaxy/issues/5142 is complete we could
+                    # at least tell the user what happened, give them a warning.
+                    log.debug(message)
+                    continue
                 output = outputs[output_name]
                 self._record_workflow_output(
                     step,

test/api/test_workflows.py

@@ -1621,7 +1621,9 @@ test_data:
         self.assertEqual("chr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\n", content)
 
     def wait_for_invocation_and_jobs(self, history_id, workflow_id, invocation_id, assert_ok=True):
-        self.workflow_populator.wait_for_invocation(workflow_id, invocation_id)
+        state = self.workflow_populator.wait_for_invocation(workflow_id, invocation_id)
+        if assert_ok:
+            assert state == "scheduled", state
         time.sleep(.5)
         self.dataset_populator.wait_for_history_jobs(history_id, assert_ok=assert_ok)
         time.sleep(.5)
@@ -1715,6 +1717,29 @@ test_data:
         content = self.dataset_populator.get_history_dataset_details(history_id, wait=True, assert_ok=True)
         assert content["name"] == "foo was replaced"
 
+    @skip_without_tool("output_filter")
+    def test_optional_workflow_output(self):
+        with self.dataset_populator.test_history() as history_id:
+            run_object = self._run_jobs("""
+class: GalaxyWorkflow
+inputs: []
+outputs:
+  - id: wf_output_1
+    source: output_filter#out_1
+steps:
+  - tool_id: output_filter
+    label: output_filter
+    state:
+      produce_out_1: False
+      filter_text_1: '1'
+test_data: {}
+    """, history_id=history_id, wait=False)
+            self.wait_for_invocation_and_jobs(history_id, run_object.workflow_id, run_object.invocation_id)
+            contents = self.__history_contents(history_id)
+            assert len(contents) == 1
+            okay_dataset = contents[0]
+            assert okay_dataset["state"] == "ok"
+
     @skip_without_tool("cat")
     def test_run_rename_collection_element(self):
         history_id = self.dataset_populator.new_history()
@@ -2424,13 +2449,18 @@ steps:
 
     def __assert_lines_hid_line_count_is(self, history, hid, lines):
         contents_url = "histories/%s/contents" % history
-        history_contents_response = self._get(contents_url)
-        self._assert_status_code_is(history_contents_response, 200)
-        hda_summary = next(hc for hc in history_contents_response.json() if hc["hid"] == hid)
+        history_contents = self.__history_contents(history)
+        hda_summary = next(hc for hc in history_contents if hc["hid"] == hid)
         hda_info_response = self._get("%s/%s" % (contents_url, hda_summary["id"]))
         self._assert_status_code_is(hda_info_response, 200)
         self.assertEqual(hda_info_response.json()["metadata_data_lines"], lines)
 
+    def __history_contents(self, history_id):
+        contents_url = "histories/%s/contents" % history_id
+        history_contents_response = self._get(contents_url)
+        self._assert_status_code_is(history_contents_response, 200)
+        return history_contents_response.json()
+
     def __invoke_workflow(self, *args, **kwds):
         return self.workflow_populator.invoke_workflow(*args, **kwds)