diff --git a/lib/galaxy/jobs/actions/post.py b/lib/galaxy/jobs/actions/post.py index 18e3c0c1e78..822fb83634c 100644 --- a/lib/galaxy/jobs/actions/post.py +++ b/lib/galaxy/jobs/actions/post.py @@ -86,11 +86,27 @@ class RenameDatasetAction(DefaultJobAction): verbose_name = "Rename Dataset" @classmethod - def execute(cls, app, sa_session, action, job, replacement_dict): + def execute_on_mapped_over(cls, app, sa_session, action, step_inputs, step_outputs, replacement_dict): # Prevent renaming a dataset to the empty string. + input_names = {} + # Lookp through inputs find one with "to_be_replaced" input + # variable name, and get the replacement name + for input_key, step_input in step_inputs.items(): + if step_input and hasattr(step_input, "name"): + input_names[input_key] = step_input.name + + new_name = cls._gen_new_name(action, input_names, replacement_dict) + if new_name: + for name, step_output in step_outputs.items(): + if action.output_name == '' or name == action.output_name: + step_output.name = new_name + + @classmethod + def _gen_new_name(self, action, input_names, replacement_dict): + new_name = None + if action.action_arguments and action.action_arguments.get('newname', ''): new_name = action.action_arguments['newname'] - # TODO: Unify and simplify replacement options. # Add interface through workflow editor UI @@ -135,20 +151,7 @@ class RenameDatasetAction(DefaultJobAction): # show correct valid inputs. input_file_var = input_file_var.replace(".", "|") - replacement = "" - # Lookp through inputs find one with "to_be_replaced" input - # variable name, and get the replacement name - for input_assoc in job.input_datasets: - if input_assoc.name == input_file_var: - replacement = input_assoc.dataset.name - - # Ditto for collections... - for input_assoc in job.input_dataset_collections: - if input_assoc.name == input_file_var: - # Either a HDCA or a DCE - only HDCA has a name. - has_collection = input_assoc.dataset_collection - if has_collection and hasattr(has_collection, "name"): - replacement = has_collection.name + replacement = input_names.get(input_file_var, "") # In case name was None. replacement = replacement or '' @@ -175,10 +178,34 @@ class RenameDatasetAction(DefaultJobAction): if replacement_dict: for k, v in replacement_dict.items(): new_name = new_name.replace("${%s}" % k, v) + + return new_name + + @classmethod + def execute(cls, app, sa_session, action, job, replacement_dict): + input_names = {} + # Lookp through inputs find one with "to_be_replaced" input + # variable name, and get the replacement name + for input_assoc in job.input_datasets: + input_names[input_assoc.name] = input_assoc.dataset.name + + # Ditto for collections... + for input_assoc in job.input_dataset_collections: + # Either a HDCA or a DCE - only HDCA has a name. + has_collection = input_assoc.dataset_collection + if has_collection and hasattr(has_collection, "name"): + input_names[input_assoc.name] = has_collection.name + + new_name = cls._gen_new_name(action, input_names, replacement_dict) + if new_name: for dataset_assoc in job.output_datasets: if action.output_name == '' or dataset_assoc.name == action.output_name: dataset_assoc.dataset.name = new_name + for dataset_collection_assoc in job.output_dataset_collection_instances: + if action.output_name == '' or dataset_collection_assoc.name == action.output_name: + dataset_collection_assoc.dataset_collection_instance.name = new_name + @classmethod def get_short_str(cls, pja): # Prevent renaming a dataset to the empty string. @@ -380,8 +407,13 @@ class ActionBox(object): 'ColumnSetAction', 'EmailAction', 'DeleteIntermediatesAction', 'TagDatasetAction', 'RemoveTagDatasetAction'] + # Actions that can be applied ahead of the job execution while workflow is still + # being scheduled and jobs created. immediate_actions = ['ChangeDatatypeAction', 'RenameDatasetAction', 'TagDatasetAction', 'RemoveTagDatasetAction'] + # Actions that will be applied to implicit mapped over collection outputs and not + # just individual outputs when steps include mapped over tools and implicit collection outputs. + mapped_over_output_actions = ['RenameDatasetAction'] @classmethod def get_short_str(cls, action): @@ -412,6 +444,11 @@ class ActionBox(object): pass return npd + @classmethod + def execute_on_mapped_over(cls, app, sa_session, pja, step_inputs, step_outputs, replacement_dict=None): + if pja.action_type in ActionBox.actions: + ActionBox.actions[pja.action_type].execute_on_mapped_over(app, sa_session, pja, step_inputs, step_outputs, replacement_dict) + @classmethod def execute(cls, app, sa_session, pja, job, replacement_dict=None): if pja.action_type in ActionBox.actions: diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index 863761d8c23..0daea4f2d0b 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -933,6 +933,11 @@ class ToolModule(WorkflowModule): step_outputs.update(execution_tracker.output_collections) progress.set_step_outputs(invocation_step, step_outputs, already_persisted=not invocation_step.is_new) + if collection_info: + step_inputs = mapping_params.param_template + step_inputs.update(collection_info.collections) + + self._handle_mapped_over_post_job_actions(step, step_inputs, step_outputs, invocation.replacement_dict) if execution_tracker.execution_errors: message = "Failed to create one or more job(s) for workflow step." raise Exception(message) @@ -971,6 +976,18 @@ class ToolModule(WorkflowModule): visit_input_values(tool.inputs, step.state.inputs, callback) return collections_to_match + def _effective_post_job_actions(self, step): + effective_post_job_actions = step.post_job_actions[:] + for key, value in self.runtime_post_job_actions.items(): + effective_post_job_actions.append(self.__to_pja(key, value, None)) + return effective_post_job_actions + + def _handle_mapped_over_post_job_actions(self, step, step_inputs, step_outputs, replacement_dict): + effective_post_job_actions = self._effective_post_job_actions(step) + for pja in effective_post_job_actions: + if pja.action_type in ActionBox.mapped_over_output_actions: + ActionBox.execute_on_mapped_over(self.trans.app, self.trans.sa_session, pja, step_inputs, step_outputs, replacement_dict) + def _handle_post_job_actions(self, step, job, replacement_dict): # Create new PJA associations with the created job, to be run on completion. # PJA Parameter Replacement (only applies to immediate actions-- rename specifically, for now) @@ -979,9 +996,7 @@ class ToolModule(WorkflowModule): # Combine workflow and runtime post job actions into the effective post # job actions for this execution. flush_required = False - effective_post_job_actions = step.post_job_actions[:] - for key, value in self.runtime_post_job_actions.items(): - effective_post_job_actions.append(self.__to_pja(key, value, None)) + effective_post_job_actions = self._effective_post_job_actions(step) for pja in effective_post_job_actions: if pja.action_type in ActionBox.immediate_actions: ActionBox.execute(self.trans.app, self.trans.sa_session, pja, job, replacement_dict) diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py index 3cadfea5e4d..18b051647ee 100644 --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -2000,7 +2000,7 @@ test_data: {} assert okay_dataset["state"] == "ok" @skip_without_tool("cat") - def test_run_rename_collection_element(self): + def test_run_rename_on_mapped_over_dataset(self): history_id = self.dataset_populator.new_history() self._run_jobs(""" class: GalaxyWorkflow @@ -2029,6 +2029,69 @@ test_data: content = self.dataset_populator.get_history_dataset_details(history_id, hid=4, wait=True, assert_ok=True) name = content["name"] assert name == "my new name", name + assert content["history_content_type"] == "dataset" + content = self.dataset_populator.get_history_collection_details(history_id, hid=3, wait=True, assert_ok=True) + name = content["name"] + assert content["history_content_type"] == "dataset_collection", content + assert name == "my new name", name + + @skip_without_tool("cat") + def test_run_rename_based_on_inputs_on_mapped_over_dataset(self): + history_id = self.dataset_populator.new_history() + self._run_jobs(""" +class: GalaxyWorkflow +inputs: + - id: input1 + type: data_collection_input + collection_type: list +steps: + - tool_id: cat + label: first_cat + state: + input1: + $link: input1 + outputs: + out_file1: + rename: "#{input1} suffix" +test_data: + input1: + type: list + name: the_dataset_list + elements: + - identifier: el1 + value: 1.fastq + type: File +""", history_id=history_id) + content = self.dataset_populator.get_history_collection_details(history_id, hid=3, wait=True, assert_ok=True) + name = content["name"] + assert content["history_content_type"] == "dataset_collection", content + assert name == "the_dataset_list suffix", name + + @skip_without_tool("collection_creates_pair") + def test_run_rename_collection_output(self): + with self.dataset_populator.test_history() as history_id: + self._run_jobs(""" +class: GalaxyWorkflow +inputs: + - id: input1 +steps: + - tool_id: collection_creates_pair + state: + input1: + $link: input1 + outputs: + paired_output: + rename: "my new name" +test_data: + input1: + value: 1.fasta + type: File + name: fasta1 +""", history_id=history_id) + details1 = self.dataset_populator.get_history_collection_details(history_id, hid=4, wait=True, assert_ok=True) + + assert details1["name"] == "my new name", details1 + assert details1["history_content_type"] == "dataset_collection" @skip_without_tool("create_2") def test_run_rename_multiple_outputs(self):