From 2b8df110d0fad7c874c35c3176bf18d65cbcdbe3 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Tue, 30 Jan 2018 10:20:54 -0500 Subject: [PATCH 1/3] Apply rename PJA to collection outputs including $ replacements. Update PJA collection test to test HDCA is renamed also. --- lib/galaxy/jobs/actions/post.py | 24 ++++++++++++++++++++++++ lib/galaxy/workflow/modules.py | 18 ++++++++++++++---- test/api/test_workflows.py | 5 +++++ 3 files changed, 43 insertions(+), 4 deletions(-) diff --git a/lib/galaxy/jobs/actions/post.py b/lib/galaxy/jobs/actions/post.py index 18e3c0c1e78..5536c1d1bd5 100644 --- a/lib/galaxy/jobs/actions/post.py +++ b/lib/galaxy/jobs/actions/post.py @@ -85,6 +85,20 @@ class RenameDatasetAction(DefaultJobAction): name = "RenameDatasetAction" verbose_name = "Rename Dataset" + @classmethod + def execute_on_mapped_over(cls, app, sa_session, action, step_outputs, replacement_dict): + # Prevent renaming a dataset to the empty string. + if action.action_arguments and action.action_arguments.get('newname', ''): + new_name = action.action_arguments['newname'] + + if replacement_dict: + for k, v in replacement_dict.items(): + new_name = new_name.replace("${%s}" % k, v) + + for name, step_output in step_outputs.items(): + if action.output_name == '' or name == action.output_name: + step_output.name = new_name + @classmethod def execute(cls, app, sa_session, action, job, replacement_dict): # Prevent renaming a dataset to the empty string. @@ -380,8 +394,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 +431,11 @@ class ActionBox(object): pass return npd + @classmethod + def execute_on_mapped_over(cls, app, sa_session, pja, 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_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..dabd99ea428 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -932,7 +932,7 @@ class ToolModule(WorkflowModule): step_outputs = dict(execution_tracker.output_datasets) step_outputs.update(execution_tracker.output_collections) progress.set_step_outputs(invocation_step, step_outputs, already_persisted=not invocation_step.is_new) - + self._handle_mapped_over_post_job_actions(step, 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 +971,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_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.immediate_actions: + ActionBox.execute_on_mapped_over(self.trans.app, self.trans.sa_session, pja, 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 +991,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..020074dc48e 100644 --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -2029,6 +2029,11 @@ 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("create_2") def test_run_rename_multiple_outputs(self): From f6f9f9a8914845c995d665dc5396ac5e43187324 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Tue, 30 Jan 2018 11:10:43 -0500 Subject: [PATCH 2/3] Extend mapped collection output rename PJAs to include # replacements. Also make sure output collections in non-mapping scenarios are also properly renamed as well. --- lib/galaxy/jobs/actions/post.py | 65 ++++++++++++++++++++------------- lib/galaxy/workflow/modules.py | 11 ++++-- test/api/test_workflows.py | 60 +++++++++++++++++++++++++++++- 3 files changed, 106 insertions(+), 30 deletions(-) diff --git a/lib/galaxy/jobs/actions/post.py b/lib/galaxy/jobs/actions/post.py index 5536c1d1bd5..822fb83634c 100644 --- a/lib/galaxy/jobs/actions/post.py +++ b/lib/galaxy/jobs/actions/post.py @@ -86,25 +86,27 @@ class RenameDatasetAction(DefaultJobAction): verbose_name = "Rename Dataset" @classmethod - def execute_on_mapped_over(cls, app, sa_session, action, step_outputs, 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. - if action.action_arguments and action.action_arguments.get('newname', ''): - new_name = action.action_arguments['newname'] - - if replacement_dict: - for k, v in replacement_dict.items(): - new_name = new_name.replace("${%s}" % k, v) + 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 execute(cls, app, sa_session, action, job, replacement_dict): - # Prevent renaming a dataset to the empty string. + 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 @@ -149,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 '' @@ -189,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. @@ -432,9 +445,9 @@ class ActionBox(object): return npd @classmethod - def execute_on_mapped_over(cls, app, sa_session, pja, step_outputs, replacement_dict=None): + 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_outputs, replacement_dict) + 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): diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index dabd99ea428..62a49fbaf0b 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -932,7 +932,12 @@ class ToolModule(WorkflowModule): step_outputs = dict(execution_tracker.output_datasets) step_outputs.update(execution_tracker.output_collections) progress.set_step_outputs(invocation_step, step_outputs, already_persisted=not invocation_step.is_new) - self._handle_mapped_over_post_job_actions(step, step_outputs, invocation.replacement_dict) + + 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) @@ -977,11 +982,11 @@ class ToolModule(WorkflowModule): 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_outputs, replacement_dict): + 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.immediate_actions: - ActionBox.execute_on_mapped_over(self.trans.app, self.trans.sa_session, pja, step_outputs, replacement_dict) + 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. diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py index 020074dc48e..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 @@ -2035,6 +2035,64 @@ test_data: 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): with self.dataset_populator.test_history() as history_id: From 0529153ac090de0601f38af2fbf4f9a4fcccb615 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Fri, 2 Feb 2018 09:54:30 -0500 Subject: [PATCH 3/3] Fix bug in PJAs for collection mapping. Reported by @mvdbeek here (https://github.com/galaxyproject/galaxy/pull/5416#issuecomment-362194157). --- lib/galaxy/workflow/modules.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index 62a49fbaf0b..0daea4f2d0b 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -985,7 +985,7 @@ class ToolModule(WorkflowModule): 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.immediate_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):