Merge pull request #16933 from martenson/merge231in

merge release_23.1 into dev

client/src/components/WorkflowInvocationState/InvocationMessage.vue

@@ -65,7 +65,12 @@ const workflow = computed(() => {
 });
 
 const workflowStep = computed(() => {
-    if ("workflow_step_id" in props.invocationMessage && workflow.value) {
+    if (
+        "workflow_step_id" in props.invocationMessage &&
+        props.invocationMessage.workflow_step_id !== undefined &&
+        props.invocationMessage.workflow_step_id !== null &&
+        workflow.value
+    ) {
         return workflow.value.steps[props.invocationMessage.workflow_step_id];
     }
     return undefined;
@@ -149,10 +154,14 @@ const infoString = computed(() => {
             invocationMessage.workflow_step_id + 1
         } is a conditional step and the result of the when expression is not a boolean type.`;
     } else if (reason === "unexpected_failure") {
-        if (invocationMessage.details) {
-            return `${failFragment} an unexpected failure occurred: '${invocationMessage.details}'`;
+        let atStep = "";
+        if (invocationMessage.workflow_step_id !== null && invocationMessage.workflow_step_id !== undefined) {
+            atStep = ` at step ${invocationMessage.workflow_step_id + 1}`;
         }
-        return `${failFragment} an unexpected failure occurred.`;
+        if (invocationMessage.details) {
+            return `${failFragment} an unexpected failure occurred${atStep}: '${invocationMessage.details}'`;
+        }
+        return `${failFragment} an unexpected failure occurred${atStep}.`;
     } else if (reason === "workflow_output_not_found") {
         return `Defined workflow output '${invocationMessage.output_name}' was not found in step ${
             invocationMessage.workflow_step_id + 1

client/src/components/WorkflowInvocationState/invocationMessageModel.ts

@@ -77,7 +77,7 @@ export interface GenericInvocationFailureCollectionFailedEncodedDatabaseIdField
     /**
      * HistoryDatasetCollectionAssociation ID that relates to failure.
      */
-    hdca_id?: string;
+    hdca_id: string;
     /**
      * Workflow step id of step that caused failure.
      */
@@ -92,7 +92,7 @@ export interface GenericInvocationFailureCollectionFailedInt {
     /**
      * HistoryDatasetCollectionAssociation ID that relates to failure.
      */
-    hdca_id?: number;
+    hdca_id: number;
     /**
      * Workflow step id of step that caused failure.
      */
@@ -159,7 +159,7 @@ export interface GenericInvocationFailureJobFailedEncodedDatabaseIdField {
     /**
      * Job ID that relates to failure.
      */
-    job_id?: string;
+    job_id: string;
     /**
      * Workflow step id of step that caused failure.
      */
@@ -174,7 +174,7 @@ export interface GenericInvocationFailureJobFailedInt {
     /**
      * Job ID that relates to failure.
      */
-    job_id?: number;
+    job_id: number;
     /**
      * Workflow step id of step that caused failure.
      */
@@ -232,6 +232,10 @@ export interface GenericInvocationUnexpectedFailureEncodedDatabaseIdField {
      * May contains details to help troubleshoot this problem.
      */
     details?: string;
+    /**
+     * Workflow step id of step that failed.
+     */
+    workflow_step_id?: number;
 }
 export interface GenericInvocationUnexpectedFailureInt {
     reason: "unexpected_failure";
@@ -239,4 +243,8 @@ export interface GenericInvocationUnexpectedFailureInt {
      * May contains details to help troubleshoot this problem.
      */
     details?: string;
+    /**
+     * Workflow step id of step that failed.
+     */
+    workflow_step_id?: number;
 }

lib/galaxy/datatypes/registry.py

@@ -600,13 +600,14 @@ class Registry:
         return self.datatypes_by_extension.get(ext, None)
 
     def change_datatype(self, data, ext):
-        data.extension = ext
-        # call init_meta and copy metadata from itself.  The datatype
-        # being converted *to* will handle any metadata copying and
-        # initialization.
-        if data.has_data():
-            data.set_size()
-            data.init_meta(copy_from=data)
+        if data.extension != ext:
+            data.extension = ext
+            # call init_meta and copy metadata from itself.  The datatype
+            # being converted *to* will handle any metadata copying and
+            # initialization.
+            if data.has_data():
+                data.set_size()
+                data.init_meta(copy_from=data)
         return data
 
     def load_datatype_converters(self, toolbox, use_cached=False):

lib/galaxy/exceptions/__init__.py

@@ -151,6 +151,16 @@ class ToolInputsNotReadyException(MessageException):
     error_code = error_codes_by_name["TOOL_INPUTS_NOT_READY"]
 
 
+class ToolInputsNotOKException(MessageException):
+    def __init__(self, err_msg=None, type="info", *, src: str, id: int, **extra_error_info):
+        super().__init__(err_msg, type, **extra_error_info)
+        self.src = src
+        self.id = id
+
+    status_code = 400
+    error_code = error_codes_by_name["TOOL_INPUTS_NOT_OK"]
+
+
 class RealUserRequiredException(MessageException):
     status_code = 400
     error_code = error_codes_by_name["REAL_USER_REQUIRED"]

lib/galaxy/exceptions/error_codes.json

@@ -94,6 +94,11 @@
         "code": 400016,
         "message": "Only real users can make this request."
     },
+    {
+        "name": "TOOL_INPUTS_NOT_OK",
+        "code": 400017,
+        "message": "Tool inputs not in required OK state."
+    },
     {
         "name": "USER_AUTHENTICATION_FAILED",
         "code": 401001,

lib/galaxy/jobs/handler.py

@@ -1198,12 +1198,16 @@ class DefaultJobDispatcher:
         except KeyError:
             log.error(f"({job_wrapper.job_id}) Invalid job runner: {runner_name}")
             job_wrapper.fail(DEFAULT_JOB_RUNNER_FAILURE_MESSAGE)
+            return None
         if get_task_runner and job_wrapper.can_split() and runner.runner_name != "PulsarJobRunner":
             return self.job_runners["tasks"]
         return runner
 
     def put(self, job_wrapper):
         runner = self.get_job_runner(job_wrapper, get_task_runner=True)
+        if runner is None:
+            # Something went wrong, we've already failed the job wrapper
+            return
         if isinstance(job_wrapper, TaskWrapper):
             # DBTODO Refactor
             log.debug(f"({job_wrapper.job_id}) Dispatching task {job_wrapper.task_id} to task runner")

lib/galaxy/metadata/set_metadata.py

@@ -448,7 +448,7 @@ def set_metadata_portable(
                     object_store_update_actions.append(
                         partial(push_if_necessary, object_store, dataset, external_filename)
                     )
-                object_store_update_actions.append(partial(reset_external_filename, dataset))
+                    object_store_update_actions.append(partial(reset_external_filename, dataset))
                 object_store_update_actions.append(partial(dataset.set_total_size))
                 object_store_update_actions.append(partial(export_store.add_dataset, dataset))
                 if dataset_instance_id not in unnamed_id_to_path:

lib/galaxy/model/metadata.py

@@ -605,7 +605,18 @@ class FileParameter(MetadataParameter):
         if isinstance(value, int):
             return session.get(galaxy.model.MetadataFile, value)
         else:
-            return session.execute(select(galaxy.model.MetadataFile).filter_by(uuid=value)).scalar_one()
+            wrapped_value = session.execute(
+                select(galaxy.model.MetadataFile).filter_by(uuid=value)
+            ).scalar_one_or_none()
+            if wrapped_value:
+                return wrapped_value
+            else:
+                # If we've simultaneously copied the  dataset and we've changed the datatype on the
+                # copy we may not have committed the MetadataFile yet, so we need to commit the session.
+                # TODO: It would be great if we can avoid the commit in the future.
+                with transaction(session):
+                    session.commit()
+            return session.execute(select(galaxy.model.MetadataFile).filter_by(uuid=value)).scalar_one_or_none()
 
     def make_copy(self, value, target_context: MetadataCollection, source_context):
         session = target_context._object_session(target_context.parent)

lib/galaxy/schema/invocation.py

@@ -51,7 +51,10 @@ class StepOrderIndexGetter(GetterDict):
         # Fetch the order_index when serializing for the API,
         # which makes much more sense when pointing to steps.
         if key == "workflow_step_id":
-            return self._obj.workflow_step.order_index
+            if self._obj.workflow_step:
+                return self._obj.workflow_step.order_index
+            else:
+                return default
         elif key == "dependent_workflow_step_id":
             if self._obj.dependent_workflow_step_id:
                 return self._obj.dependent_workflow_step.order_index
@@ -132,6 +135,7 @@ class GenericInvocationFailureWhenNotBoolean(InvocationFailureMessageBase[Databa
 class GenericInvocationUnexpectedFailure(InvocationMessageBase, Generic[DatabaseIdT]):
     reason: Literal[FailureReason.unexpected_failure]
     details: Optional[str] = Field(None, description="May contains details to help troubleshoot this problem.")
+    workflow_step_id: Optional[int] = Field(None, description="Workflow step id of step that failed.")
 
 
 class GenericInvocationWarning(InvocationMessageBase, Generic[DatabaseIdT]):

lib/galaxy/tools/__init__.py

@@ -39,7 +39,10 @@ from galaxy import (
     exceptions,
     model,
 )
-from galaxy.exceptions import ToolInputsNotReadyException
+from galaxy.exceptions import (
+    ToolInputsNotOKException,
+    ToolInputsNotReadyException,
+)
 from galaxy.job_execution import output_collect
 from galaxy.metadata import get_metadata_compute_strategy
 from galaxy.model import (
@@ -2875,7 +2878,7 @@ class ExpressionTool(Tool):
                         copy_object = input_dataset
                         break
                 if copy_object is None:
-                    raise Exception("Failed to find dataset output.")
+                    raise exceptions.MessageException("Failed to find dataset output.")
                 out_data[key].copy_from(copy_object)
 
     def parse_environment_variables(self, tool_source):
@@ -3217,8 +3220,10 @@ class DatabaseOperationTool(Tool):
 
             if self.require_dataset_ok:
                 if state != model.Dataset.states.OK:
-                    raise ValueError(
-                        f"Tool requires inputs to be in valid state, but dataset {input_dataset} is in state '{input_dataset.state}'"
+                    raise ToolInputsNotOKException(
+                        f"Tool requires inputs to be in valid state, but dataset {input_dataset} is in state '{input_dataset.state}'",
+                        src="hda",
+                        id=input_dataset.id,
                     )
 
         for input_dataset in input_datasets.values():
@@ -3347,7 +3352,7 @@ class ExtractDatasetCollectionTool(DatabaseOperationTool):
         elif how == "by_index":
             extracted_element = collection[int(incoming["which"]["index"])]
         else:
-            raise Exception("Invalid tool parameters.")
+            raise exceptions.MessageException("Invalid tool parameters.")
         extracted = extracted_element.element_object
         extracted_o = extracted.copy(
             copy_tags=extracted.tags, new_name=extracted_element.element_identifier, flush=False
@@ -3387,7 +3392,9 @@ class MergeCollectionTool(DatabaseOperationTool):
                 if element_identifier not in identifiers_map:
                     identifiers_map[element_identifier] = []
                 elif dupl_actions == "fail":
-                    raise Exception(f"Duplicate collection element identifiers found for [{element_identifier}]")
+                    raise exceptions.MessageException(
+                        f"Duplicate collection element identifiers found for [{element_identifier}]"
+                    )
                 identifiers_map[element_identifier].append(input_num)
 
         for copy, input_list in enumerate(input_lists):
@@ -3579,12 +3586,12 @@ class SortTool(DatabaseOperationTool):
                 except KeyError:
                     hdca_history_name = f"{hdca.hid}: {hdca.name}"
                     message = f"List of element identifiers does not match element identifiers in collection '{hdca_history_name}'"
-                    raise Exception(message)
+                    raise exceptions.MessageException(message)
             else:
                 message = f"Number of lines must match number of list elements ({len(elements)}), but file has {data_lines} lines"
-                raise Exception(message)
+                raise exceptions.MessageException(message)
         else:
-            raise Exception(f"Unknown sort_type '{sorttype}'")
+            raise exceptions.MessageException(f"Unknown sort_type '{sorttype}'")
 
         if presort_elements is not None:
             sorted_elements = [x[1] for x in sorted(presort_elements, key=lambda x: x[0])]
@@ -3616,7 +3623,7 @@ class RelabelFromFileTool(DatabaseOperationTool):
         def add_copied_value_to_new_elements(new_label, dce_object):
             new_label = new_label.strip()
             if new_label in new_elements:
-                raise Exception(
+                raise exceptions.MessageException(
                     f"New identifier [{new_label}] appears twice in resulting collection, these values must be unique."
                 )
             if getattr(dce_object, "history_content_type", None) == "dataset":
@@ -3629,7 +3636,7 @@ class RelabelFromFileTool(DatabaseOperationTool):
         with open(new_labels_path) as fh:
             new_labels = fh.readlines(1024 * 1000000)
         if strict and len(hdca.collection.elements) != len(new_labels):
-            raise Exception("Relabel mapping file contains incorrect number of identifiers")
+            raise exceptions.MessageException("Relabel mapping file contains incorrect number of identifiers")
         if how_type == "tabular":
             # We have a tabular file, where the first column is an existing element identifier,
             # and the second column is the new element identifier.
@@ -3641,7 +3648,7 @@ class RelabelFromFileTool(DatabaseOperationTool):
                 default = None if strict else element_identifier
                 new_label = new_labels_dict.get(element_identifier, default)
                 if not new_label:
-                    raise Exception(f"Failed to find new label for identifier [{element_identifier}]")
+                    raise exceptions.MessageException(f"Failed to find new label for identifier [{element_identifier}]")
                 add_copied_value_to_new_elements(new_label, dce_object)
         else:
             # If new_labels_dataset_assoc is not a two-column tabular dataset we label with the current line of the dataset
@@ -3650,7 +3657,7 @@ class RelabelFromFileTool(DatabaseOperationTool):
                 add_copied_value_to_new_elements(new_labels[i], dce_object)
         for key in new_elements.keys():
             if not re.match(r"^[\w\- \.,]+$", key):
-                raise Exception(f"Invalid new collection identifier [{key}]")
+                raise exceptions.MessageException(f"Invalid new collection identifier [{key}]")
         self._add_datasets_to_history(history, new_elements.values())
         output_collections.create_collection(
             next(iter(self.outputs.values())), "output", elements=new_elements, propagate_hda_tags=False

lib/galaxy/tools/execute.py

@@ -19,6 +19,7 @@ from typing import (
 from boltons.iterutils import remap
 
 from galaxy import model
+from galaxy.exceptions import ToolInputsNotOKException
 from galaxy.model.base import transaction
 from galaxy.model.dataset_collections.matching import MatchingCollections
 from galaxy.model.dataset_collections.structure import (
@@ -143,14 +144,17 @@ def execute(
     if check_inputs_ready:
         for params in execution_tracker.param_combinations:
             # This will throw an exception if the tool is not ready.
-            check_inputs_ready(
-                tool,
-                trans,
-                params,
-                history,
-                execution_cache=execution_cache,
-                collection_info=collection_info,
-            )
+            try:
+                check_inputs_ready(
+                    tool,
+                    trans,
+                    params,
+                    history,
+                    execution_cache=execution_cache,
+                    collection_info=collection_info,
+                )
+            except ToolInputsNotOKException as e:
+                execution_tracker.record_error(e)
 
     execution_tracker.ensure_implicit_collections_populated(history, mapping_params.param_template)
     job_count = len(execution_tracker.param_combinations)

lib/galaxy/webapps/galaxy/controllers/history.py

@@ -490,6 +490,10 @@ class HistoryController(BaseUIController, SharableMixin, UsesAnnotations, UsesIt
             history_is_current = history_to_view == trans.history
         else:
             history_to_view = trans.history
+            if not history_to_view:
+                raise exceptions.RequestParameterMissingException(
+                    "No 'id' parameter provided for history, and user does not have a current history."
+                )
             user_is_owner = True
             history_is_current = True
 

lib/galaxy/workflow/modules.py

@@ -2314,6 +2314,16 @@ class ToolModule(WorkflowModule):
         if execution_tracker.execution_errors:
             # TODO: formalize into InvocationFailure ?
             message = f"Failed to create {len(execution_tracker.execution_errors)} job(s) for workflow step {step.order_index + 1}: {str(execution_tracker.execution_errors[0])}"
+            for error in execution_tracker.execution_errors:
+                # try first to raise a structured invocation error message
+                if isinstance(error, exceptions.ToolInputsNotOKException) and error.src == "hda":
+                    raise FailWorkflowEvaluation(
+                        why=InvocationFailureDatasetFailed(
+                            reason=FailureReason.dataset_failed,
+                            hda_id=error.id,
+                            workflow_step_id=step.id,
+                        )
+                    )
             raise exceptions.MessageException(message)
 
         return complete

lib/galaxy/workflow/run.py

@@ -240,12 +240,20 @@ class WorkflowInvoker:
             except modules.DelayedWorkflowEvaluation as de:
                 step_delayed = delayed_steps = True
                 self.progress.mark_step_outputs_delayed(step, why=de.why)
-            except Exception:
+            except Exception as e:
                 log.exception(
                     "Failed to schedule %s, problem occurred on %s.",
                     self.workflow_invocation.workflow.log_str(),
                     step.log_str(),
                 )
+                if isinstance(e, MessageException):
+                    # This is the highest level at which we can inject the step id
+                    # to provide some more context to the exception.
+                    raise modules.FailWorkflowEvaluation(
+                        why=InvocationUnexpectedFailure(
+                            reason=FailureReason.unexpected_failure, details=str(e), workflow_step_id=step.id
+                        )
+                    )
                 raise
 
             if not step_delayed:

lib/galaxy_test/api/test_workflows.py

@@ -4060,6 +4060,78 @@ outputs:
         assert "workflow_step_id" in message
         assert message["output_name"] == "does_not_exist"
 
+    @skip_without_tool("__APPLY_RULES__")
+    @skip_without_tool("job_properties")
+    def test_workflow_failed_input_not_ok(self, history_id):
+        summary = self._run_workflow(
+            """
+class: GalaxyWorkflow
+steps:
+  job_props:
+    tool_id: job_properties
+    state:
+      thebool: true
+      failbool: true
+  apply:
+    tool_id: __APPLY_RULES__
+    in:
+      input: job_props/list_output
+    state:
+      rules:
+        rules:
+          - type: add_column_metadata
+            value: identifier0
+        mapping:
+          - type: list_identifiers
+            columns: [0]
+        """,
+            history_id=history_id,
+            assert_ok=False,
+            wait=True,
+        )
+        invocation_details = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
+        assert invocation_details["state"] == "failed"
+        assert len(invocation_details["messages"]) == 1
+        message = invocation_details["messages"][0]
+        assert message["reason"] == "dataset_failed"
+        assert message["workflow_step_id"] == 1
+
+    @skip_without_tool("__RELABEL_FROM_FILE__")
+    def test_workflow_failed_with_message_exception(self, history_id):
+        summary = self._run_workflow(
+            """
+class: GalaxyWorkflow
+inputs:
+  input_collection:
+    collection_type: list
+    type: collection
+  relabel_file:
+    type: data
+steps:
+  relabel:
+    tool_id: __RELABEL_FROM_FILE__
+    in:
+      input: input_collection
+      how|labels: relabel_file
+test_data:
+  input_collection:
+    collection_type: "list:list"
+  relabel_file:
+    value: 1.bed
+    type: File
+        """,
+            history_id=history_id,
+            assert_ok=False,
+            wait=True,
+        )
+        invocation_details = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
+        assert invocation_details["state"] == "failed"
+        assert len(invocation_details["messages"]) == 1
+        message = invocation_details["messages"][0]
+        assert message["reason"] == "unexpected_failure"
+        assert message["workflow_step_id"] == 2
+        assert "Invalid new collection identifier" in message["details"]
+
     @skip_without_tool("identifier_multiple")
     def test_invocation_map_over(self, history_id):
         summary = self._run_workflow(

test/integration/test_upload_configuration_options.py

@@ -30,6 +30,7 @@ from typing import (
 
 from requests import Response
 
+from galaxy.tool_util.verify.test_data import TestDataResolver
 from galaxy.util.unittest import TestCase
 from galaxy_test.base.api_util import TEST_USER
 from galaxy_test.base.constants import (
@@ -75,17 +76,19 @@ class BaseUploadContentConfigurationInstance(integration_util.IntegrationInstanc
         response = self.dataset_populator.fetch(payload, assert_ok=assert_ok, wait=wait)
         return response
 
-    def _write_file(self, dir_path: str, content: str, filename: str = "test") -> str:
-        """Helper for writing ftp/server dir files."""
-        self._ensure_directory(dir_path)
-        path = os.path.join(dir_path, filename)
-        with open(path, "w") as f:
-            f.write(content)
-        return path
 
-    def _ensure_directory(self, path: str) -> None:
-        if not os.path.exists(path):
-            os.makedirs(path)
+def _write_file(dir_path: str, content: str, filename: str = "test") -> str:
+    """Helper for writing ftp/server dir files."""
+    _ensure_directory(dir_path)
+    path = os.path.join(dir_path, filename)
+    with open(path, "w") as f:
+        f.write(content)
+    return path
+
+
+def _ensure_directory(path: str) -> None:
+    if not os.path.exists(path):
+        os.makedirs(path)
 
 
 class BaseUploadContentConfigurationTestCase(BaseUploadContentConfigurationInstance, TestCase):
@@ -190,6 +193,42 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase):
         assert os.path.exists(path)
 
 
+def fetch_link_data_only(
+    library_populator: LibraryPopulator, dataset_populator: DatasetPopulator, test_data_resolver: TestDataResolver
+):
+    history_id, library, destination = library_populator.setup_fetch_to_folder("fetch_and_link")
+    bed_test_data_path = test_data_resolver.get_filename("4.bed")
+    assert os.path.exists(bed_test_data_path)
+    items = [{"src": "path", "path": bed_test_data_path, "info": "my cool bed", "link_data_only": True}]
+    targets = [{"destination": destination, "items": items}]
+    payload = {
+        "history_id": history_id,  # TODO: Shouldn't be needed :(
+        "targets": targets,
+    }
+    dataset_populator.fetch(payload)
+    dataset = library_populator.get_library_contents_with_path(library["id"], "/4.bed")
+    assert dataset["file_size"] == 61, dataset
+    assert dataset["file_name"] == bed_test_data_path, dataset
+    assert os.path.exists(bed_test_data_path)
+
+
+def link_data_only(server_dir: str, library_populator: LibraryPopulator) -> None:
+    content = "hello world\n"
+    dir_path = os.path.join(server_dir, "lib1")
+    file_path = _write_file(dir_path, content)
+    library = library_populator.new_private_library("serverdirupload")
+    # upload $GALAXY_ROOT/test-data/library
+    payload, files = library_populator.create_dataset_request(
+        library, upload_option="upload_directory", server_dir="lib1", link_data=True
+    )
+    response = library_populator.raw_library_contents_create(library["id"], payload, files=files)
+    assert response.status_code == 200, response.json()
+    dataset = response.json()[0]
+    ok_dataset = library_populator.wait_on_library_dataset(library["id"], dataset["id"])
+    assert ok_dataset["file_size"] == 12, ok_dataset
+    assert ok_dataset["file_name"] == file_path, ok_dataset
+
+
 class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase):
     require_admin_user = True
 
@@ -506,9 +545,9 @@ class TestEnableFtpPurgeUploadConfiguration(BaseFtpUploadConfigurationTestCase):
 class TestAdvancedFtpUploadFetch(BaseFtpUploadConfigurationTestCase):
     def test_fetch_ftp_directory(self, history_id: str) -> None:
         dir_path = self._get_user_ftp_path()
-        self._write_file(os.path.join(dir_path, "subdir"), "content 1", filename="1")
-        self._write_file(os.path.join(dir_path, "subdir"), "content 22", filename="2")
-        self._write_file(os.path.join(dir_path, "subdir"), "content 333", filename="3")
+        _write_file(os.path.join(dir_path, "subdir"), "content 1", filename="1")
+        _write_file(os.path.join(dir_path, "subdir"), "content 22", filename="2")
+        _write_file(os.path.join(dir_path, "subdir"), "content 333", filename="3")
         target = {
             "destination": {"type": "hdca"},
             "elements_from": "directory",
@@ -525,9 +564,9 @@ class TestAdvancedFtpUploadFetch(BaseFtpUploadConfigurationTestCase):
 
     def test_fetch_nested_elements_from(self, history_id: str) -> None:
         dir_path = self._get_user_ftp_path()
-        self._write_file(os.path.join(dir_path, "subdir1"), "content 1", filename="1")
-        self._write_file(os.path.join(dir_path, "subdir1"), "content 22", filename="2")
-        self._write_file(os.path.join(dir_path, "subdir2"), "content 333", filename="3")
+        _write_file(os.path.join(dir_path, "subdir1"), "content 1", filename="1")
+        _write_file(os.path.join(dir_path, "subdir1"), "content 22", filename="2")
+        _write_file(os.path.join(dir_path, "subdir2"), "content 333", filename="3")
         elements = [
             {
                 "name": "subdirel1",
@@ -667,7 +706,7 @@ class TestUploadOptionsFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase
         shutil.copyfile(input_path, os.path.join(target_dir, test_data_path))
 
     def _write_user_ftp_file(self, path: str, content: str) -> str:
-        return self._write_file(os.path.join(self.ftp_dir(), TEST_USER), content, filename=path)
+        return _write_file(os.path.join(self.ftp_dir(), TEST_USER), content, filename=path)
 
 
 class TestServerDirectoryOffByDefault(BaseUploadContentConfigurationTestCase):
@@ -721,20 +760,7 @@ class TestServerDirectoryValidUsage(BaseUploadContentConfigurationTestCase):
         assert library_dataset["file_size"] == 12, library_dataset
 
     def test_link_data_only(self) -> None:
-        content = "hello world\n"
-        dir_path = os.path.join(self.server_dir(), "lib1")
-        file_path = self._write_file(dir_path, content)
-        library = self.library_populator.new_private_library("serverdirupload")
-        # upload $GALAXY_ROOT/test-data/library
-        payload, files = self.library_populator.create_dataset_request(
-            library, upload_option="upload_directory", server_dir="lib1", link_data=True
-        )
-        response = self.library_populator.raw_library_contents_create(library["id"], payload, files=files)
-        assert response.status_code == 200, response.json()
-        dataset = response.json()[0]
-        ok_dataset = self.library_populator.wait_on_library_dataset(library["id"], dataset["id"])
-        assert ok_dataset["file_size"] == 12, ok_dataset
-        assert ok_dataset["file_name"] == file_path, ok_dataset
+        link_data_only(self.server_dir(), self.library_populator)
 
     @classmethod
     def server_dir(cls) -> str:
@@ -813,20 +839,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase):
         assert os.path.exists(bed_test_data_path)
 
     def test_fetch_link_data_only(self) -> None:
-        history_id, library, destination = self.library_populator.setup_fetch_to_folder("fetch_and_link")
-        bed_test_data_path = self.test_data_resolver.get_filename("4.bed")
-        assert os.path.exists(bed_test_data_path)
-        items = [{"src": "path", "path": bed_test_data_path, "info": "my cool bed", "link_data_only": True}]
-        targets = [{"destination": destination, "items": items}]
-        payload = {
-            "history_id": history_id,  # TODO: Shouldn't be needed :(
-            "targets": targets,
-        }
-        self.dataset_populator.fetch(payload)
-        dataset = self.library_populator.get_library_contents_with_path(library["id"], "/4.bed")
-        assert dataset["file_size"] == 61, dataset
-        assert dataset["file_name"] == bed_test_data_path, dataset
-        assert os.path.exists(bed_test_data_path)
+        fetch_link_data_only(self.library_populator, self.dataset_populator, self.test_data_resolver)
 
     def test_fetch_recursive_archive(self) -> None:
         history_id, library, destination = self.library_populator.setup_fetch_to_folder("recursive_archive")
@@ -948,3 +961,26 @@ class TestDirectoryAndCompressedTypes(BaseUploadContentConfigurationTestCase):
             history_id=history_id,
         )
         self.dataset_populator.wait_for_job(response["jobs"][0]["id"])
+
+
+class TestLinkDataUploadExtendedMetadata(BaseUploadContentConfigurationTestCase):
+    require_admin_user = True
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config) -> None:
+        super().handle_galaxy_config_kwds(config)
+        config["allow_path_paste"] = True
+        config["metadata_strategy"] = "extended"
+        server_dir = cls.server_dir()
+        os.makedirs(server_dir)
+        config["library_import_dir"] = server_dir
+
+    @classmethod
+    def server_dir(cls) -> str:
+        return cls.temp_config_dir("server")
+
+    def test_fetch_link_data_only(self) -> None:
+        fetch_link_data_only(self.library_populator, self.dataset_populator, self.test_data_resolver)
+
+    def test_link_data_only(self) -> None:
+        link_data_only(self.server_dir(), self.library_populator)

tools/filters/sorter.py

@@ -41,7 +41,8 @@ def main():
 
         # grep comments
         grep_comments = ["grep", "^#"]
-        exit_code = subprocess.call(grep_comments, stdout=output_fh)
+        exit_code = subprocess.call(grep_comments, stdin=input_fh, stdout=output_fh)
+        input_fh.seek(0)
         if exit_code not in [0, 1]:
             stop_err("Searching for comment lines failed")