mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
John Chilton
8af243c176
Allow the rules DSL & GUI component to operate on existing collections to allow filtering, sorting, modifying identifiers and general re-organization of existing collections (e.g. the outputs of tools). Implementing this as a collection operation tool so that it should be executable interactively in the tool form and in a batch fashion as part of workflow executions. For this to be tracked properly as a tool execution and to work properly in the tool form, I've implemented a new tool framework and tool form parameter type called "rules". This can thought of as a more GUI friendly alternative to my proposed collection operations that consumed JavaScript expressions. This includes API tests for both tool and workflow execution of the new tool as well as Selenium tests for tool form execution, workflow editor interactions, and workflow running.
379 lines
16 KiB
Python
379 lines
16 KiB
Python
import json
|
|
|
|
from base.workflow_fixtures import (
|
|
WORKFLOW_NESTED_SIMPLE,
|
|
WORKFLOW_SIMPLE_CAT_TWICE,
|
|
WORKFLOW_SIMPLE_MAPPING,
|
|
WORKFLOW_WITH_INVALID_STATE,
|
|
WORKFLOW_WITH_OLD_TOOL_VERSION,
|
|
WORKFLOW_WITH_OUTPUT_COLLECTION,
|
|
WORKFLOW_WITH_RULES_1,
|
|
WORKFLOW_WITH_RULES_2,
|
|
)
|
|
|
|
from .framework import (
|
|
retry_assertion_during_transitions,
|
|
selenium_test,
|
|
SeleniumTestCase
|
|
)
|
|
|
|
|
|
class WorkflowEditorTestCase(SeleniumTestCase):
|
|
|
|
ensure_registered = True
|
|
|
|
@selenium_test
|
|
def test_basics(self):
|
|
editor = self.components.workflow_editor
|
|
|
|
name = self.workflow_create_new()
|
|
edit_name_element = self.components.workflow_editor.edit_name.wait_for_visible()
|
|
assert name in edit_name_element.text, edit_name_element.text
|
|
|
|
editor.canvas_body.wait_for_visible()
|
|
editor.tool_menu.wait_for_visible()
|
|
|
|
self.screenshot("workflow_editor_blank")
|
|
|
|
self.components._.left_panel_drag.wait_for_visible()
|
|
self.components._.left_panel_collapse.wait_for_and_click()
|
|
|
|
self.sleep_for(self.wait_types.UX_RENDER)
|
|
|
|
self.screenshot("workflow_editor_left_collapsed")
|
|
|
|
self.components._.right_panel_drag.wait_for_visible()
|
|
self.components._.right_panel_collapse.wait_for_and_click()
|
|
|
|
self.sleep_for(self.wait_types.UX_RENDER)
|
|
|
|
self.screenshot("workflow_editor_left_and_right_collapsed")
|
|
|
|
@selenium_test
|
|
def test_data_input(self):
|
|
editor = self.components.workflow_editor
|
|
|
|
name = self.workflow_create_new()
|
|
editor.canvas_body.wait_for_visible()
|
|
editor.tool_menu.wait_for_visible()
|
|
editor.tool_menu_section_link(section_name="inputs").wait_for_and_click()
|
|
editor.tool_menu_item_link(section_name="inputs", item_name="data_input").wait_for_and_click()
|
|
self.screenshot("workflow_editor_data_input_new")
|
|
editor.label_input.wait_for_and_send_keys("input1")
|
|
editor.annotation_input.wait_for_and_send_keys("my cool annotation")
|
|
editor.label_input.wait_for_and_click() # Seems to help force the save of whole annotation.
|
|
self.sleep_for(self.wait_types.UX_RENDER)
|
|
self.screenshot("workflow_editor_data_input_filled_in")
|
|
self.workflow_editor_save_and_close()
|
|
self.workflow_index_open_with_name(name)
|
|
data_input_node = editor.node._(label="input1")
|
|
data_input_node.title.wait_for_and_click()
|
|
|
|
label = editor.label_input.wait_for_value()
|
|
assert label == "input1", label
|
|
assert editor.annotation_input.wait_for_value() == "my cool annotation"
|
|
|
|
data_input_node.destroy.wait_for_and_click()
|
|
data_input_node.wait_for_absent()
|
|
self.screenshot("workflow_editor_data_input_deleted")
|
|
|
|
@selenium_test
|
|
def test_collection_input(self):
|
|
editor = self.components.workflow_editor
|
|
|
|
name = self.workflow_create_new()
|
|
editor.canvas_body.wait_for_visible()
|
|
editor.tool_menu.wait_for_visible()
|
|
editor.tool_menu_section_link(section_name="inputs").wait_for_and_click()
|
|
editor.tool_menu_item_link(section_name="inputs", item_name="data_collection_input").wait_for_and_click()
|
|
self.screenshot("workflow_editor_data_input_collection_new")
|
|
editor.label_input.wait_for_and_send_keys("input1")
|
|
editor.annotation_input.wait_for_and_send_keys("my cool annotation")
|
|
editor.label_input.wait_for_and_click() # Seems to help force the save of whole annotation.
|
|
self.sleep_for(self.wait_types.UX_RENDER)
|
|
self.screenshot("workflow_editor_data_input_collection_filled_in")
|
|
self.workflow_editor_save_and_close()
|
|
self.workflow_index_open_with_name(name)
|
|
data_input_node = editor.node._(label="input1")
|
|
data_input_node.title.wait_for_and_click()
|
|
|
|
label = editor.label_input.wait_for_value()
|
|
assert label == "input1", label
|
|
assert editor.annotation_input.wait_for_value() == "my cool annotation"
|
|
|
|
data_input_node.destroy.wait_for_and_click()
|
|
data_input_node.wait_for_absent()
|
|
self.screenshot("workflow_editor_data_input_deleted")
|
|
|
|
@selenium_test
|
|
def test_existing_connections(self):
|
|
self.open_in_workflow_editor(WORKFLOW_SIMPLE_CAT_TWICE)
|
|
|
|
editor = self.components.workflow_editor
|
|
self.assert_connected("input1#output", "first_cat#input1")
|
|
self.screenshot("workflow_editor_connection_simple")
|
|
|
|
cat_node = editor.node._(label="first_cat")
|
|
cat_input = cat_node.input_terminal(name="input1")
|
|
cat_input.wait_for_and_click()
|
|
editor.connector_destroy_callout.wait_for_visible()
|
|
self.screenshot("workflow_editor_connection_callout")
|
|
editor.connector_destroy_callout.wait_for_and_click()
|
|
self.assert_not_connected("input1#output", "first_cat#input1")
|
|
self.screenshot("workflow_editor_connection_destroyed")
|
|
|
|
self.workflow_editor_connect("input1#output", "first_cat#input1", screenshot_partial="workflow_editor_connection_dragging")
|
|
self.assert_connected("input1#output", "first_cat#input1")
|
|
|
|
@selenium_test
|
|
def test_rendering_output_collection_connections(self):
|
|
self.open_in_workflow_editor(WORKFLOW_WITH_OUTPUT_COLLECTION)
|
|
self.workflow_editor_maximize_center_pane()
|
|
self.screenshot("workflow_editor_output_collections")
|
|
|
|
@selenium_test
|
|
def test_simple_mapping_connections(self):
|
|
self.open_in_workflow_editor(WORKFLOW_SIMPLE_MAPPING)
|
|
self.workflow_editor_maximize_center_pane()
|
|
self.screenshot("workflow_editor_simple_mapping")
|
|
self.assert_connected("input1#output", "cat#input1")
|
|
self.assert_input_mapped("cat#input1")
|
|
self.workflow_editor_destroy_connection("cat#input1")
|
|
self.assert_input_not_mapped("cat#input1")
|
|
self.assert_not_connected("input1#output", "cat#input1")
|
|
self.workflow_editor_connect("input1#output", "cat#input1")
|
|
self.assert_input_mapped("cat#input1")
|
|
|
|
@selenium_test
|
|
def test_rendering_simple_nested_workflow(self):
|
|
self.open_in_workflow_editor(WORKFLOW_NESTED_SIMPLE)
|
|
self.workflow_editor_maximize_center_pane()
|
|
self.screenshot("workflow_editor_simple_nested")
|
|
|
|
@selenium_test
|
|
def test_rendering_rules_workflow_1(self):
|
|
self.open_in_workflow_editor(WORKFLOW_WITH_RULES_1)
|
|
rule_output = "apply#output"
|
|
random_lines_input = "random_lines#input"
|
|
|
|
self.workflow_editor_maximize_center_pane()
|
|
self.screenshot("workflow_editor_rules_1")
|
|
self.assert_connected(rule_output, random_lines_input)
|
|
self.assert_input_mapped(random_lines_input)
|
|
self.workflow_editor_destroy_connection(random_lines_input)
|
|
self.assert_not_connected(rule_output, random_lines_input)
|
|
self.assert_input_not_mapped(random_lines_input)
|
|
self.workflow_editor_connect(rule_output, random_lines_input)
|
|
self.assert_connected(rule_output, random_lines_input)
|
|
self.assert_input_mapped(random_lines_input)
|
|
|
|
@selenium_test
|
|
def test_rendering_rules_workflow_2(self):
|
|
self.open_in_workflow_editor(WORKFLOW_WITH_RULES_2)
|
|
self.workflow_editor_maximize_center_pane(collapse_right=False)
|
|
|
|
editor = self.components.workflow_editor
|
|
rule_builder = self.components.rule_builder
|
|
|
|
rule_output = "apply#output"
|
|
copy_list_input = "copy_list#input1"
|
|
|
|
apply_node = editor.node._(label="apply")
|
|
|
|
self.assert_connected(rule_output, copy_list_input)
|
|
self.assert_input_mapped(copy_list_input)
|
|
self.workflow_editor_destroy_connection(copy_list_input)
|
|
self.assert_not_connected(rule_output, copy_list_input)
|
|
self.assert_input_not_mapped(copy_list_input)
|
|
self.workflow_editor_connect(rule_output, copy_list_input)
|
|
self.assert_connected(rule_output, copy_list_input)
|
|
self.assert_input_mapped(copy_list_input)
|
|
|
|
apply_node.title.wait_for_and_click()
|
|
self.screenshot("workflow_editor_rules_2_form")
|
|
self.tool_parameter_edit_rules()
|
|
rule_builder._.wait_for_visible()
|
|
self.screenshot("workflow_editor_rules_2_builder")
|
|
new_rules = dict(
|
|
rules=[{"type": "add_column_metadata", "value": "identifier0"}],
|
|
mapping=[{"type": "list_identifiers", "columns": [0]}],
|
|
)
|
|
self.rule_builder_set_source(json.dumps(new_rules))
|
|
rule_builder.main_button_ok.wait_for_and_click()
|
|
apply_node.title.wait_for_and_click()
|
|
self.sleep_for(self.wait_types.UX_RENDER)
|
|
# screenshot should have async warning about connection removed
|
|
self.screenshot("workflow_editor_rules_2_after_change")
|
|
self.assert_input_not_mapped(copy_list_input)
|
|
# changing output collection type remove outbound connections, so this
|
|
# this needs to be re-connected. Remove this re-connection if we upgrade
|
|
# the workflow editor to try to re-establish the connection with different
|
|
# mapping.
|
|
self.workflow_editor_connect(rule_output, copy_list_input)
|
|
self.assert_connected(rule_output, copy_list_input)
|
|
# Regardless - this rules now say to connect a list to a list instead of a list
|
|
# to a list:list, so there should be no mapping anymore even after connected.
|
|
self.assert_input_not_mapped(copy_list_input)
|
|
|
|
@selenium_test
|
|
def test_save_as(self):
|
|
name = self.workflow_upload_yaml_with_random_name(WORKFLOW_SIMPLE_CAT_TWICE)
|
|
self.workflow_index_open()
|
|
self.workflow_index_open_with_name(name)
|
|
self.sleep_for(self.wait_types.UX_RENDER)
|
|
self.screenshot("workflow_editor_edit_menu")
|
|
self.workflow_editor_click_option("Save As")
|
|
|
|
@selenium_test
|
|
def test_editor_tool_upgrade_message(self):
|
|
workflow_populator = self.workflow_populator
|
|
workflow_populator.upload_yaml_workflow(WORKFLOW_WITH_OLD_TOOL_VERSION, exact_tools=True)
|
|
self.workflow_index_open()
|
|
self.workflow_index_click_option("Edit")
|
|
self.assert_modal_has_text("Using version '0.2' instead of version '0.0.1'")
|
|
self.screenshot("workflow_editor_tool_upgrade")
|
|
|
|
@selenium_test
|
|
def test_editor_invalid_tool_state(self):
|
|
workflow_populator = self.workflow_populator
|
|
workflow_populator.upload_yaml_workflow(WORKFLOW_WITH_INVALID_STATE, exact_tools=True)
|
|
self.workflow_index_open()
|
|
self.workflow_index_click_option("Edit")
|
|
self.assert_modal_has_text("Using version '0.2' instead of version '0.0.1'")
|
|
self.assert_modal_has_text("Using default: '1'")
|
|
self.screenshot("workflow_editor_invalid_state")
|
|
|
|
@selenium_test
|
|
def test_missing_tools(self):
|
|
workflow_populator = self.workflow_populator
|
|
workflow_populator.upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
steps:
|
|
- tool_id: missing
|
|
label: first_cat
|
|
state:
|
|
foo: bar
|
|
""")
|
|
self.workflow_index_open()
|
|
self.workflow_index_click_option("Edit")
|
|
self.assert_modal_has_text("Tool is not installed")
|
|
self.screenshot("workflow_editor_missing_tool")
|
|
|
|
def workflow_editor_save_and_close(self):
|
|
self.workflow_editor_click_option("Save")
|
|
self.workflow_editor_click_option("Close")
|
|
|
|
def workflow_editor_maximize_center_pane(self, collapse_left=True, collapse_right=True):
|
|
if collapse_left:
|
|
self.components._.left_panel_collapse.wait_for_and_click()
|
|
if collapse_right:
|
|
self.components._.right_panel_collapse.wait_for_and_click()
|
|
self.sleep_for(self.wait_types.UX_RENDER)
|
|
|
|
def workflow_editor_connect(self, source, sink, screenshot_partial=None):
|
|
source_id, sink_id = self.workflow_editor_source_sink_terminal_ids(source, sink)
|
|
source_element = self.driver.find_element_by_css_selector("#" + source_id)
|
|
sink_element = self.driver.find_element_by_css_selector("#" + sink_id)
|
|
|
|
ac = self.action_chains()
|
|
ac = ac.move_to_element(source_element).click_and_hold()
|
|
if screenshot_partial:
|
|
ac = ac.move_to_element_with_offset(sink_element, -5, 0)
|
|
ac.perform()
|
|
self.sleep_for(self.wait_types.UX_RENDER)
|
|
self.screenshot(screenshot_partial)
|
|
ac = self.action_chains()
|
|
|
|
ac = ac.move_to_element(sink_element).release().perform()
|
|
|
|
def assert_connected(self, source, sink):
|
|
source_id, sink_id = self.workflow_editor_source_sink_terminal_ids(source, sink)
|
|
self.components.workflow_editor.connector_for(source_id=source_id, sink_id=sink_id).wait_for_visible()
|
|
|
|
def assert_not_connected(self, source, sink):
|
|
source_id, sink_id = self.workflow_editor_source_sink_terminal_ids(source, sink)
|
|
self.components.workflow_editor.connector_for(source_id=source_id, sink_id=sink_id).wait_for_absent()
|
|
|
|
def open_in_workflow_editor(self, yaml_content):
|
|
name = self.workflow_upload_yaml_with_random_name(yaml_content)
|
|
self.workflow_index_open()
|
|
self.workflow_index_open_with_name(name)
|
|
self.workflow_editor_click_option("Auto Re-layout")
|
|
|
|
def workflow_editor_source_sink_terminal_ids(self, source, sink):
|
|
editor = self.components.workflow_editor
|
|
|
|
source_node_label, source_output = source.split("#", 1)
|
|
sink_node_label, sink_input = sink.split("#", 1)
|
|
|
|
source_node = editor.node._(label=source_node_label)
|
|
sink_node = editor.node._(label=sink_node_label)
|
|
|
|
source_node.wait_for_visible()
|
|
sink_node.wait_for_visible()
|
|
|
|
output_terminal = source_node.output_terminal(name=source_output)
|
|
input_terminal = sink_node.input_terminal(name=sink_input)
|
|
|
|
output_element = output_terminal.wait_for_visible()
|
|
input_element = input_terminal.wait_for_visible()
|
|
|
|
source_id = output_element.get_attribute("id")
|
|
sink_id = input_element.get_attribute("id")
|
|
|
|
return source_id, sink_id
|
|
|
|
def workflow_editor_destroy_connection(self, sink):
|
|
editor = self.components.workflow_editor
|
|
|
|
sink_node_label, sink_input_name = sink.split("#", 1)
|
|
sink_node = editor.node._(label=sink_node_label)
|
|
sink_input = sink_node.input_terminal(name=sink_input_name)
|
|
sink_input.wait_for_and_click()
|
|
editor.connector_destroy_callout.wait_for_and_click()
|
|
|
|
def assert_input_mapped(self, sink):
|
|
editor = self.components.workflow_editor
|
|
sink_node_label, sink_input_name = sink.split("#", 1)
|
|
sink_node = editor.node._(label=sink_node_label)
|
|
sink_mapping_icon = sink_node.input_mapping_icon(name=sink_input_name)
|
|
sink_mapping_icon.wait_for_visible()
|
|
|
|
def assert_input_not_mapped(self, sink):
|
|
editor = self.components.workflow_editor
|
|
sink_node_label, sink_input_name = sink.split("#", 1)
|
|
sink_node = editor.node._(label=sink_node_label)
|
|
sink_mapping_icon = sink_node.input_mapping_icon(name=sink_input_name)
|
|
sink_mapping_icon.wait_for_absent_or_hidden()
|
|
|
|
def workflow_index_open_with_name(self, name):
|
|
self.workflow_index_open()
|
|
self.workflow_index_search_for(name)
|
|
self.workflow_index_click_option("Edit")
|
|
|
|
def workflow_upload_yaml_with_random_name(self, content):
|
|
workflow_populator = self.workflow_populator
|
|
name = self._get_random_name()
|
|
workflow_populator.upload_yaml_workflow(content, name=name)
|
|
return name
|
|
|
|
def workflow_create_new(self, annotation=None):
|
|
self.workflow_index_open()
|
|
self.click_button_new_workflow()
|
|
form_element = self.driver.find_element_by_id("submit")
|
|
name = self._get_random_name()
|
|
annotation = annotation or self._get_random_name()
|
|
inputs = self.driver.find_elements_by_class_name("ui-input")
|
|
inputs[0].send_keys(name)
|
|
inputs[1].send_keys(annotation)
|
|
form_element.click()
|
|
return name
|
|
|
|
@retry_assertion_during_transitions
|
|
def assert_modal_has_text(self, expected_text):
|
|
modal_element = self.wait_for_selector_visible(self.modal_body_selector())
|
|
text = modal_element.text
|
|
assert expected_text in text, "Failed to find expected text [%s] in modal text [%s]" % (expected_text, text)
|