diff --git a/test/integration/test_workflow_invocation.py b/test/integration/test_workflow_invocation.py index 5ac1fc680f8a..72276bf77979 100644 --- a/test/integration/test_workflow_invocation.py +++ b/test/integration/test_workflow_invocation.py @@ -7,6 +7,48 @@ from galaxy_test.base.uses_shed_api import UsesShedApi from galaxy_test.driver import integration_util +OPTIONAL_INPUT_BOOL_WORKFLOW = """ +class: GalaxyWorkflow +inputs: + optional_data: + optional: true + type: data +steps: + optional_data_to_boolean: + tool_id: toolshed.g2.bx.psu.edu/repos/iuc/map_param_value/map_param_value/0.2.0 + tool_state: + input_param_type: + mappings: + - from: null + to: false + type: data + output_param_type: boolean + unmapped: + default_value: true + on_unmapped: default + in: + input_param_type|input_param: optional_data + optional_to_non_optional: + tool_id: pick_value + tool_state: + style_cond: + pick_style: first + type_cond: + param_type: data + in: + style_cond|type_cond|pick_from_0|value: + source: optional_data + cat1: + tool_id: cat1 + when: $(inputs.when) + in: + input1: optional_to_non_optional/data_param + when: optional_data_to_boolean/output_param_boolean +outputs: + cat1: + outputSource: cat1/out_file1 +""" + class TestWorkflowInvocation(integration_util.IntegrationTestCase, UsesShedApi): dataset_populator: DatasetPopulator @@ -18,6 +60,33 @@ def setUp(self): self.dataset_populator = DatasetPopulator(self.galaxy_interactor) self.workflow_populator = WorkflowPopulator(self.galaxy_interactor) + def test_run_workflow_optional_data_skips_step(self) -> None: + self.install_repository("iuc", "map_param_value", "5ac8a4bf7a8d") + with self.dataset_populator.test_history() as history_id: + summary = self.workflow_populator.run_workflow(OPTIONAL_INPUT_BOOL_WORKFLOW, history_id=history_id) + invocation_details = self.workflow_populator.get_invocation(summary.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_optional_data_provided_runs_step(self) -> None: + self.install_repository("iuc", "map_param_value", "5ac8a4bf7a8d") + with self.dataset_populator.test_history() as history_id: + summary = self.workflow_populator.run_workflow( + OPTIONAL_INPUT_BOOL_WORKFLOW, + test_data={ + "optional_data": { + "value": "1.bed", + "type": "File", + } + }, + history_id=history_id, + ) + invocation_details = self.workflow_populator.get_invocation(summary.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"] == "ok") == 1, step["jobs"] + def test_run_workflow_with_missing_tool(self): self.install_repository("iuc", "compose_text_param", "feb3acba1e0a") # 0.1.0 with self.dataset_populator.test_history() as history_id: