Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy_test.selenium.test_workflow_run
import yaml
from galaxy_test.base import rules_test_data
from galaxy_test.base.populators import load_data_dict
from galaxy_test.base.workflow_fixtures import (
WORKFLOW_NESTED_REPLACEMENT_PARAMETER,
WORKFLOW_NESTED_RUNTIME_PARAMETER,
WORKFLOW_NESTED_SIMPLE,
WORKFLOW_RENAME_ON_REPLACEMENT_PARAM,
WORKFLOW_RUNTIME_PARAMETER_SIMPLE,
WORKFLOW_SELECT_FROM_OPTIONAL_DATASET,
WORKFLOW_SIMPLE_CAT_TWICE,
WORKFLOW_WITH_CUSTOM_REPORT_1,
WORKFLOW_WITH_CUSTOM_REPORT_1_TEST_DATA,
WORKFLOW_WITH_DYNAMIC_OUTPUT_COLLECTION,
WORKFLOW_WITH_OLD_TOOL_VERSION,
WORKFLOW_WITH_RULES_1,
)
from .framework import (
managed_history,
selenium_test,
SeleniumTestCase,
UsesHistoryItemAssertions,
)
[docs]class WorkflowRunTestCase(SeleniumTestCase, UsesHistoryItemAssertions):
ensure_registered = True
[docs] @selenium_test
@managed_history
def test_simple_execution(self):
self.perform_upload(self.get_filename("1.fasta"))
self.wait_for_history()
self.open_in_workflow_run(WORKFLOW_SIMPLE_CAT_TWICE)
self.screenshot("workflow_run_simple_ready")
self.workflow_run_submit()
self.sleep_for(self.wait_types.UX_TRANSITION)
self.screenshot("workflow_run_simple_submitted")
self.history_panel_wait_for_hid_ok(2, allowed_force_refreshes=1)
self.history_panel_click_item_title(hid=2, wait=True)
self.assert_item_summary_includes(2, "2 sequences")
self.screenshot("workflow_run_simple_complete")
[docs] @selenium_test
@managed_history
def test_runtime_parameters_simple(self):
self.perform_upload(self.get_filename("1.txt"))
self.wait_for_history()
self.open_in_workflow_run(WORKFLOW_RUNTIME_PARAMETER_SIMPLE)
self.tool_parameter_div("num_lines")
self.screenshot("workflow_run_runtime_parameters_initial")
self._set_num_lines_to_3("num_lines")
self.screenshot("workflow_run_runtime_parameters_modified")
self.workflow_run_submit()
self._assert_has_3_lines_after_run(hid=2)
[docs] @selenium_test
@managed_history
def test_subworkflows_simple(self):
self.perform_upload(self.get_filename("1.txt"))
self.wait_for_history()
self.open_in_workflow_run(WORKFLOW_NESTED_SIMPLE)
self.components.workflow_run.subworkflow_step_icon.wait_for_visible()
self.screenshot("workflow_run_nested_collapsed")
self.components.workflow_run.subworkflow_step_icon.wait_for_and_click()
self.screenshot("workflow_run_nested_open")
[docs] @selenium_test
@managed_history
def test_subworkflow_runtime_parameters(self):
self.perform_upload(self.get_filename("1.txt"))
self.wait_for_history()
self.open_in_workflow_run(WORKFLOW_NESTED_RUNTIME_PARAMETER)
self.components.workflow_run.subworkflow_step_icon.wait_for_visible()
self.screenshot("workflow_run_nested_parameters_collapsed")
self.components.workflow_run.subworkflow_step_icon.wait_for_and_click()
self.screenshot("workflow_run_nested_parameters_open")
self._set_num_lines_to_3("1|num_lines")
self.workflow_run_submit()
self._assert_has_3_lines_after_run(hid=2)
[docs] @selenium_test
@managed_history
def test_replacement_parameters(self):
self.perform_upload(self.get_filename("1.txt"))
self.wait_for_history()
self.open_in_workflow_run(WORKFLOW_RENAME_ON_REPLACEMENT_PARAM)
self.screenshot("workflow_run_rename_simple_empty")
self._set_replacement_parameter("replaceme", "moocow")
self.screenshot("workflow_run_rename_simple_input")
self.workflow_run_submit()
output_hid = 2
self.history_panel_wait_for_hid_ok(output_hid, allowed_force_refreshes=1)
history_id = self.current_history_id()
details = self.dataset_populator.get_history_dataset_details(history_id, hid=output_hid)
assert details["name"] == "moocow suffix", details
[docs] @selenium_test
@managed_history
def test_step_parameter_inputs(self):
self.perform_upload(self.get_filename("1.txt"))
self.wait_for_history()
self.open_in_workflow_run("""
class: GalaxyWorkflow
inputs:
input_int: integer
input_data: data
steps:
simple_constructs:
tool_id: simple_constructs
label: tool_exec
in:
inttest: input_int
files_0|file: input_data
""")
workflow_run = self.components.workflow_run
input_div_element = workflow_run.input_div(label="input_int").wait_for_visible()
input_element = input_div_element.find_element_by_css_selector("input")
input_element.clear()
input_element.send_keys("12345")
self.screenshot("workflow_run_step_parameter_input")
self.workflow_run_submit()
output_hid = 2
self.history_panel_wait_for_hid_ok(output_hid, allowed_force_refreshes=1)
history_id = self.current_history_id()
content = self.dataset_populator.get_history_dataset_content(history_id, hid=output_hid)
assert "12345" in content, content
assert "chr6_hla_hap2" in content
[docs] @selenium_test
@managed_history
def test_replacement_parameters_on_subworkflows(self):
self.perform_upload(self.get_filename("1.txt"))
self.wait_for_history()
self.open_in_workflow_run(WORKFLOW_NESTED_REPLACEMENT_PARAMETER)
self.screenshot("workflow_run_rename_subworkflow_empty")
self._set_replacement_parameter("replaceme", "moocow")
self.screenshot("workflow_run_rename_subworkflow_input")
self.workflow_run_submit()
output_hid = 2
self.history_panel_wait_for_hid_ok(output_hid, allowed_force_refreshes=1)
history_id = self.current_history_id()
details = self.dataset_populator.get_history_dataset_details(history_id, hid=output_hid)
assert details["name"] == "moocow suffix", details
[docs] @selenium_test
def test_execution_with_tool_upgrade(self):
name = self.workflow_upload_yaml_with_random_name(WORKFLOW_WITH_OLD_TOOL_VERSION, exact_tools=True)
self.workflow_run_with_name(name)
self.sleep_for(self.wait_types.UX_TRANSITION)
# Check that this tool form contains a warning about different versions.
self.assert_message(self.components.workflow_run.warning, contains="different versions")
self.screenshot("workflow_run_tool_upgrade")
[docs] @selenium_test
@managed_history
def test_execution_with_multiple_inputs(self):
history_id, inputs = self.workflow_run_setup_inputs(WORKFLOW_WITH_DYNAMIC_OUTPUT_COLLECTION)
self.open_in_workflow_run(WORKFLOW_WITH_DYNAMIC_OUTPUT_COLLECTION)
self.workflow_run_specify_inputs(inputs)
self.screenshot("workflow_run_two_inputs")
self.workflow_run_submit()
self.history_panel_wait_for_hid_ok(7, allowed_force_refreshes=1)
content = self.dataset_populator.get_history_dataset_content(history_id, hid=7)
self.assertEqual("10.0\n30.0\n20.0\n40.0\n", content)
[docs] @selenium_test
@managed_history
def test_execution_with_rules(self):
history_id, inputs = self.workflow_run_setup_inputs(WORKFLOW_WITH_RULES_1)
self.open_in_workflow_run(WORKFLOW_WITH_RULES_1)
self.workflow_run_specify_inputs(inputs)
self.screenshot("workflow_run_rules")
self.workflow_run_submit()
self.history_panel_wait_for_hid_ok(6, allowed_force_refreshes=1)
output_content = self.dataset_populator.get_history_collection_details(history_id, hid=6)
rules_test_data.check_example_2(output_content, self.dataset_populator)
[docs] @selenium_test
@managed_history
def test_execution_with_custom_invocation_report(self):
history_id, inputs = self.workflow_run_setup_inputs(WORKFLOW_WITH_CUSTOM_REPORT_1_TEST_DATA)
self.open_in_workflow_run(WORKFLOW_WITH_CUSTOM_REPORT_1)
self.workflow_run_specify_inputs(inputs)
self.workflow_run_submit()
self.sleep_for(self.wait_types.UX_TRANSITION)
self.screenshot("workflow_run_invocation_report")
self.workflow_populator.wait_for_history_workflows(history_id, expected_invocation_count=1)
invocation_0 = self.workflow_populator.history_invocations(history_id)[0]
self.get("workflows/invocations/report?id=%s" % invocation_0["id"])
self.wait_for_selector_visible(".embedded-dataset")
self.screenshot("workflow_report_custom_1")
[docs] @selenium_test
@managed_history
def test_execution_with_null_optional_select_from_data(self):
self.open_in_workflow_run(WORKFLOW_SELECT_FROM_OPTIONAL_DATASET)
self.workflow_run_submit()
history_id = self.current_history_id()
self.workflow_populator.wait_for_history_workflows(history_id, expected_invocation_count=1)
[docs] def open_in_workflow_run(self, yaml_content):
name = self.workflow_upload_yaml_with_random_name(yaml_content)
self.workflow_run_with_name(name)
[docs] def workflow_run_setup_inputs(self, content):
history_id = self.current_history_id()
yaml_content = yaml.safe_load(content)
if "test_data" in yaml_content:
test_data = yaml_content["test_data"]
else:
test_data = yaml_content
inputs, _, _ = load_data_dict(history_id, test_data, self.dataset_populator, self.dataset_collection_populator)
self.dataset_populator.wait_for_history(history_id)
return history_id, inputs
[docs] def workflow_run_specify_inputs(self, inputs):
workflow_run = self.components.workflow_run
for label, value in inputs.items():
input_div_element = workflow_run.input_data_div(label=label).wait_for_visible()
self.select2_set_value(input_div_element, "%d: " % value["hid"])
[docs] def workflow_run_with_name(self, name):
self.workflow_index_open()
self.workflow_index_search_for(name)
self.workflow_click_option(".workflow-run")
def _assert_has_3_lines_after_run(self, hid):
self.history_panel_wait_for_hid_ok(hid, allowed_force_refreshes=1)
history_id = self.current_history_id()
content = self.dataset_populator.get_history_dataset_content(history_id, hid=hid)
assert len([x for x in content.split("\n") if x]) == 3, content
def _set_num_lines_to_3(self, tour_id):
# for random_lines num_lines parameter as runtime parameter in workflow form.
div = self.tool_parameter_div(tour_id)
input_element = div.find_element_by_css_selector("input")
# runtime parameters not being set to tool default value:
# https://github.com/galaxyproject/galaxy/pull/7157
# initial_value = input_element.get_attribute("value")
# assert initial_value == "1", initial_value
input_element.clear()
input_element.send_keys("3")
def _set_replacement_parameter(self, tour_id, value):
# for random_lines num_lines parameter as runtime parameter in workflow form.
div = self.tool_parameter_div(tour_id)
input_element = div.find_element_by_css_selector("input")
initial_value = input_element.get_attribute("value")
assert initial_value == "", initial_value
input_element.clear()
input_element.send_keys(value)