[21.09] fix JobWrapper: use correct fail method

lib/galaxy/jobs/__init__.py

@@ -1696,7 +1696,13 @@ def fail(message=job.info, exception=None):
                     else:
                         # Prior to fail we need to set job.state
                         job.set_state(final_job_state)
-                        return self.fail(f"Job {job.id}'s output dataset(s) could not be read")
+                        return fail(f"Job {job.id}'s output dataset(s) could not be read")
+        else:
+            # check existence of outputs (tools may delete outputs)
+            for dataset_path in self.get_output_fnames():
+                if not os.path.exists(dataset_path.real_path):
+                    job.set_state(final_job_state)
+                    return fail(f"Job {job.id}'s output dataset(s) could not be read")
 
         job_context = ExpressionContext(dict(stdout=job.stdout, stderr=job.stderr))
         if extended_metadata:
@@ -2005,6 +2011,8 @@ def compute_outputs(self):
 
         results = []
         for da in job.output_datasets + job.output_library_datasets:
+            if da.purged:
+                continue
             da_false_path = dataset_path_rewriter.rewrite_dataset_path(da.dataset, 'output')
             mutable = da.dataset.dataset.external_filename is None
             dataset_path = DatasetPath(da.dataset.dataset.id, da.dataset.file_name, false_path=da_false_path, mutable=mutable)

lib/galaxy/metadata/set_metadata.py

@@ -273,7 +273,7 @@ def set_meta(new_dataset_instance, file_dict):
             external_filename = unnamed_id_to_path.get(dataset_instance_id, dataset_filename_override)
             if not os.path.exists(external_filename):
                 matches = glob.glob(external_filename)
-                assert len(matches) == 1, f"More than one file matched by output glob '{external_filename}'"
+                assert len(matches) == 1, f"{len(matches)} file matched by output glob '{external_filename}', should be 1"
                 external_filename = matches[0]
                 assert safe_contains(tool_job_working_directory, external_filename), f"Cannot collect output '{external_filename}' from outside of working directory"
                 created_from_basename = os.path.relpath(external_filename, os.path.join(tool_job_working_directory, 'working'))

test/functional/tools/samples_tool_conf.xml

@@ -214,6 +214,8 @@
   <tool file="interactivetool_simple.xml" />
   <tool file="interactivetool_two_entry_points.xml" />
   <tool file="converter_target_datatype.xml" />
+
+  <tool file="tool_deleting_output.xml" />
 
   <!-- Tools interesting only for building up test workflows. -->
 

test/functional/tools/tool_deleting_output.xml

@@ -0,0 +1,26 @@
+<tool id="tool_deleting_output" name="tool_deleting_output" version="0.1.0" license="AFL-3.0" profile="21.09">
+    <command><![CDATA[
+rm '$out_file1' || true && ## ensure that this does not fail.
+exit $exit_code
+    ]]></command>
+    <inputs>
+        <param name="exit_code" type="integer" value="0"/>
+    </inputs>
+    <outputs>
+        <data name="out_file1" format="txt" />
+    </outputs>
+    <tests>
+        <test expect_exit_code="0" expect_failure="true">
+            <param name="exit_code" value="0"/>
+        </test>
+        <test expect_exit_code="1" expect_failure="true">
+            <param name="exit_code" value="1"/>
+        </test>
+    </tests>
+    <help><![CDATA[
+        This tool tests what happens if the underlying program removes the output data set:
+        the job should fail independent of the exit code and the exit code should be correctly checked.
+
+        With outputs_to_working_directory this provoked https://github.com/galaxyproject/galaxy/issues/14206
+    ]]></help>
+</tool>

test/integration/test_job_outputs_to_working_directory.py

@@ -15,4 +15,4 @@ def handle_galaxy_config_kwds(cls, config):
 
 instance = integration_util.integration_module_instance(JobOutputsToWorkingDirectoryIntegrationInstance)
 
-test_tools = integration_util.integration_tool_runner(["output_format", "output_empty_work_dir", "collection_creates_pair_from_work_dir"])
+test_tools = integration_util.integration_tool_runner(["output_format", "output_empty_work_dir", "collection_creates_pair_from_work_dir", "tool_deleting_output"])