Loading lib/galaxy_test/api/test_workflows.py +78 −0 Original line number Diff line number Diff line Loading @@ -2157,6 +2157,84 @@ should_run: if step["workflow_step_label"] == "a_tool_step": assert sum(1 for j in step["jobs"] if j["state"] == "skipped") == 1 def test_run_nested_conditional_workflow_steps(self): with self.dataset_populator.test_history() as history_id: summary = self._run_workflow( """ class: GalaxyWorkflow inputs: dataset: type: data when: type: boolean outputs: output: outputSource: outer_subworkflow/output steps: - label: outer_subworkflow when: $(inputs.when) in: dataset: source: dataset when: source: when run: class: GalaxyWorkflow label: subworkflow cat1 inputs: dataset: type: data outputs: output: outputSource: cat1_workflow/output steps: - label: cat1_workflow in: dataset: source: dataset run: class: GalaxyWorkflow label: cat1 inputs: dataset: type: data outputs: output: outputSource: cat1/out_file1 steps: - tool_id: cat1 label: cat1 in: input1: source: dataset """, test_data=""" dataset: value: 1.bed type: File when: value: false type: raw """, history_id=history_id, wait=True, assert_ok=True, ) invocation_details = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True) subworkflow_invocation_id = invocation_details["steps"][-1]["subworkflow_invocation_id"] self.workflow_populator.wait_for_invocation_and_jobs( history_id=history_id, workflow_id="whatever", invocation_id=subworkflow_invocation_id ) invocation_details = self.workflow_populator.get_invocation(subworkflow_invocation_id, step_details=True) subworkflow_invocation_id = invocation_details["steps"][-1]["subworkflow_invocation_id"] self.workflow_populator.wait_for_invocation_and_jobs( history_id=history_id, workflow_id="whatever", invocation_id=subworkflow_invocation_id ) invocation_details = self.workflow_populator.get_invocation(subworkflow_invocation_id, step_details=True) for step in invocation_details["steps"]: if step["workflow_step_label"] == "cat1": assert sum(1 for j in step["jobs"] if j["state"] == "skipped") == 1 def test_run_workflow_conditional_step_map_over_expression_tool(self): with self.dataset_populator.test_history() as history_id: summary = self._run_workflow( Loading Loading
lib/galaxy_test/api/test_workflows.py +78 −0 Original line number Diff line number Diff line Loading @@ -2157,6 +2157,84 @@ should_run: if step["workflow_step_label"] == "a_tool_step": assert sum(1 for j in step["jobs"] if j["state"] == "skipped") == 1 def test_run_nested_conditional_workflow_steps(self): with self.dataset_populator.test_history() as history_id: summary = self._run_workflow( """ class: GalaxyWorkflow inputs: dataset: type: data when: type: boolean outputs: output: outputSource: outer_subworkflow/output steps: - label: outer_subworkflow when: $(inputs.when) in: dataset: source: dataset when: source: when run: class: GalaxyWorkflow label: subworkflow cat1 inputs: dataset: type: data outputs: output: outputSource: cat1_workflow/output steps: - label: cat1_workflow in: dataset: source: dataset run: class: GalaxyWorkflow label: cat1 inputs: dataset: type: data outputs: output: outputSource: cat1/out_file1 steps: - tool_id: cat1 label: cat1 in: input1: source: dataset """, test_data=""" dataset: value: 1.bed type: File when: value: false type: raw """, history_id=history_id, wait=True, assert_ok=True, ) invocation_details = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True) subworkflow_invocation_id = invocation_details["steps"][-1]["subworkflow_invocation_id"] self.workflow_populator.wait_for_invocation_and_jobs( history_id=history_id, workflow_id="whatever", invocation_id=subworkflow_invocation_id ) invocation_details = self.workflow_populator.get_invocation(subworkflow_invocation_id, step_details=True) subworkflow_invocation_id = invocation_details["steps"][-1]["subworkflow_invocation_id"] self.workflow_populator.wait_for_invocation_and_jobs( history_id=history_id, workflow_id="whatever", invocation_id=subworkflow_invocation_id ) invocation_details = self.workflow_populator.get_invocation(subworkflow_invocation_id, step_details=True) for step in invocation_details["steps"]: if step["workflow_step_label"] == "cat1": assert sum(1 for j in step["jobs"] if j["state"] == "skipped") == 1 def test_run_workflow_conditional_step_map_over_expression_tool(self): with self.dataset_populator.test_history() as history_id: summary = self._run_workflow( Loading