import json
import yaml
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.keys import Keys
from galaxy_test.base.workflow_fixtures import (
WORKFLOW_NESTED_SIMPLE,
WORKFLOW_OPTIONAL_TRUE_INPUT_COLLECTION,
WORKFLOW_SELECT_FROM_OPTIONAL_DATASET,
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,
retry_during_transitions,
RunsWorkflows,
selenium_test,
SeleniumTestCase,
)
[docs]class WorkflowEditorTestCase(SeleniumTestCase, RunsWorkflows):
ensure_registered = True
[docs] @selenium_test
def test_basics(self):
editor = self.components.workflow_editor
annotation = "basic_test"
name = self.workflow_create_new(annotation=annotation)
self.assert_wf_name_is(name)
self.assert_wf_annotation_is(annotation)
editor.canvas_body.wait_for_visible()
editor.tool_menu.wait_for_visible()
# shouldn't have changes on fresh load
save_button = self.components.workflow_editor.save_button
save_button.wait_for_visible()
assert save_button.has_class("disabled")
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")
[docs] @selenium_test
def test_edit_annotation(self):
editor = self.components.workflow_editor
annotation = "new_annotation_test"
name = self.workflow_create_new(annotation=annotation)
edit_annotation = self.components.workflow_editor.edit_annotation
self.assert_wf_annotation_is(annotation)
editor.canvas_body.wait_for_visible()
new_annotation = "look new annotation"
edit_annotation.wait_for_and_send_keys(new_annotation)
self.assert_workflow_has_changes_and_save()
self.workflow_index_open_with_name(name)
self.assert_wf_annotation_is(new_annotation)
[docs] @selenium_test
def test_edit_name(self):
editor = self.components.workflow_editor
name = self.workflow_create_new()
editor.canvas_body.wait_for_visible()
new_name = self._get_random_name()
edit_name = self.components.workflow_editor.edit_name
edit_name.wait_for_and_send_keys(new_name)
self.assert_workflow_has_changes_and_save()
self.workflow_index_open_with_name(new_name)
self.assert_wf_name_is(name)
[docs] @selenium_test
def test_optional_select_data_field(self):
editor = self.components.workflow_editor
workflow_id = self.workflow_populator.upload_yaml_workflow(WORKFLOW_SELECT_FROM_OPTIONAL_DATASET)
self.workflow_index_open()
self.workflow_index_click_option("Edit")
editor = self.components.workflow_editor
node = editor.node._(label="select_from_dataset_optional")
node.title.wait_for_and_click()
self.components.tool_form.parameter_checkbox(parameter="select_single").wait_for_and_click()
# External (selenium-side) debounce hack for old backbone input
# TODO: remove when form elements are all converted.
self.components.tool_form.parameter_input(parameter="select_single").wait_for_and_send_keys("parameter valu")
self.sleep_for(self.wait_types.UX_RENDER)
self.components.tool_form.parameter_input(parameter="select_single").wait_for_and_send_keys("e")
self.sleep_for(self.wait_types.UX_RENDER)
self.assert_workflow_has_changes_and_save()
workflow = self.workflow_populator.download_workflow(workflow_id)
tool_state = json.loads(workflow["steps"]["0"]["tool_state"])
assert tool_state["select_single"] == "parameter value"
# Disable optional button, resets value to null
self.components.tool_form.parameter_checkbox(parameter="select_single").wait_for_and_click()
self.assert_workflow_has_changes_and_save()
workflow = self.workflow_populator.download_workflow(workflow_id)
tool_state = json.loads(workflow["steps"]["0"]["tool_state"])
assert tool_state["select_single"] is None
# Enable button but don't provide a value
self.components.tool_form.parameter_checkbox(parameter="select_single").wait_for_and_click()
self.assert_workflow_has_changes_and_save()
workflow = self.workflow_populator.download_workflow(workflow_id)
tool_state = json.loads(workflow["steps"]["0"]["tool_state"])
assert tool_state["select_single"] == ""
[docs] @selenium_test
def test_non_data_connections(self):
self.open_in_workflow_editor(
"""
class: GalaxyWorkflow
inputs:
input_int: integer
steps:
simple_constructs:
tool_id: simple_constructs
label: tool_exec
in:
inttest: input_int
cat1:
# regression test, ensures connecting works in the presence of data input terminals
tool_id: cat1
"""
)
self.screenshot("workflow_editor_parameter_connection_simple")
self.assert_connected("input_int#output", "tool_exec#inttest")
editor = self.components.workflow_editor
tool_node = editor.node._(label="tool_exec")
tool_input = tool_node.input_terminal(name="inttest")
tool_input.wait_for_and_click()
editor.connector_destroy_callout.wait_for_and_click()
self.assert_not_connected("input_int#output", "tool_exec#inttest")
self.screenshot("workflow_editor_parameter_connection_destroyed")
# When connected, cannot turn it into a RuntimeValue..
collapse_input = editor.collapse_icon(name="inttest")
collapse_input.wait_for_absent_or_hidden()
# If it is disconnected, then can specify as RuntimeValue
connect_icon = editor.connect_icon(name="inttest")
connect_icon.wait_for_visible()
connect_icon.wait_for_and_click()
collapse_input.wait_for_visible()
# Also the connector should disappear
tool_input.wait_for_absent_or_hidden()
# Now make it connected again and watch the requests
connect_icon.wait_for_and_click()
tool_input.wait_for_visible()
collapse_input.wait_for_absent_or_hidden()
self.workflow_editor_connect(
"input_int#output", "tool_exec#inttest", screenshot_partial="workflow_editor_parameter_connection_dragging"
)
self.assert_connected("input_int#output", "tool_exec#inttest")
[docs] @selenium_test
def test_non_data_map_over_carried_through(self):
# Use auto_layout=false, which prevents placing any
# step outside of the scroll area
# xref: https://github.com/galaxyproject/galaxy/issues/13211
self.open_in_workflow_editor(
"""
class: GalaxyWorkflow
inputs:
input_collection:
type: collection
collection_type: "list"
steps:
param_value_from_file:
tool_id: param_value_from_file
in:
input1: input_collection
text_input_step:
tool_id: param_text_option
in:
text_param: param_value_from_file/text_param
collection_input:
tool_id: identifier_collection
""",
auto_layout=False,
)
self.workflow_editor_connect("text_input_step#out_file1", "collection_input#input1")
self.assert_connected("text_input_step#out_file1", "collection_input#input1")
[docs] def test_connecting_display_in_upload_false_connections(self):
self.open_in_workflow_editor(
"""
class: GalaxyWorkflow
steps:
step1:
tool_id: test_sam_to_bam_conversions
step2:
tool_id: test_sam_to_bam_conversions
"""
)
self.workflow_editor_connect("step1#qname_input_sorted_bam_output", "step2#input5")
self.assert_connected("step1#qname_input_sorted_bam_output", "step2#input5")
[docs] @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")
[docs] @selenium_test
def test_reconnecting_nodes(self):
name = self.open_in_workflow_editor(WORKFLOW_SIMPLE_CAT_TWICE)
self.assert_connected("input1#output", "first_cat#input1")
self.workflow_editor_destroy_connection("first_cat#input1")
self.assert_not_connected("input1#output", "first_cat#input1")
self.workflow_editor_connect("input1#output", "first_cat#input1")
self.assert_connected("input1#output", "first_cat#input1")
self.sleep_for(self.wait_types.UX_RENDER)
self.assert_workflow_has_changes_and_save()
self.workflow_index_open_with_name(name)
self.assert_connected("input1#output", "first_cat#input1")
[docs] @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")
[docs] @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")
[docs] @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")
[docs] @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)
[docs] @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)
[docs] @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")
[docs] @staticmethod
def set_text_element(element, value):
# Try both, no harm here
element.wait_for_and_send_keys(Keys.CONTROL, "a")
element.wait_for_and_send_keys(Keys.COMMAND, "a")
element.wait_for_and_send_keys(Keys.BACKSPACE)
element.wait_for_and_send_keys(value)
[docs] @selenium_test
def test_change_datatype(self):
self.open_in_workflow_editor(
"""
class: GalaxyWorkflow
inputs: []
steps:
- tool_id: create_2
label: create_2
- tool_id: checksum
label: checksum
in:
input: create_2/out_file1
"""
)
editor = self.components.workflow_editor
self.assert_connected("create_2#out_file1", "checksum#input")
node = editor.node._(label="create_2")
node.wait_for_and_click()
editor.configure_output(output="out_file1").wait_for_and_click()
editor.change_datatype.wait_for_and_click()
editor.select_dataype_text_search.wait_for_and_send_keys("bam")
editor.select_datatype(datatype="bam").wait_for_and_click()
editor.node.output_data_row(output_name="out_file1", extension="bam").wait_for_visible()
self.assert_not_connected("create_2#out_file1", "checksum#input")
[docs] @selenium_test
def test_change_datatype_post_job_action_lost_regression(self):
self.open_in_workflow_editor(
"""
class: GalaxyWorkflow
inputs: []
steps:
- tool_id: create_2
label: create_2
outputs:
out_file1:
change_datatype: bam
- tool_id: metadata_bam
label: metadata_bam
in:
input_bam: create_2/out_file1
"""
)
self.assert_connected("create_2#out_file1", "metadata_bam#input_bam")
editor = self.components.workflow_editor
node = editor.node._(label="create_2")
node.wait_for_and_click()
self.assert_connected("create_2#out_file1", "metadata_bam#input_bam")
[docs] @selenium_test
def test_change_datatype_in_subworkflow(self):
self.open_in_workflow_editor(
"""
class: GalaxyWorkflow
inputs: []
steps:
nested_workflow:
run:
class: GalaxyWorkflow
inputs: []
steps:
- tool_id: create_2
label: create_2
outputs:
out_file1:
change_datatype: bam
outputs:
workflow_output:
outputSource: create_2/out_file1
metadata_bam:
tool_id: metadata_bam
"""
)
editor = self.components.workflow_editor
node = editor.node._(label="nested_workflow")
node.wait_for_and_click()
node.output_data_row(output_name="workflow_output", extension="bam").wait_for_visible()
# Move canvas, so terminals are in viewport
self.move_center_of_canvas(xoffset=100, yoffset=100)
self.workflow_editor_connect("nested_workflow#workflow_output", "metadata_bam#input_bam")
self.assert_connected("nested_workflow#workflow_output", "metadata_bam#input_bam")
[docs] @selenium_test
def test_editor_duplicate_node(self):
workflow_id = self.workflow_populator.upload_yaml_workflow(WORKFLOW_SIMPLE_CAT_TWICE)
self.workflow_index_open()
self.workflow_index_click_option("Edit")
editor = self.components.workflow_editor
cat_node = editor.node._(label="first_cat")
cat_node.wait_for_and_click()
self.set_text_element(editor.label_input, "source label")
# Select node using new label, ensures labels are synced between side panel and node
cat_node = editor.node._(label="source label")
self.assert_workflow_has_changes_and_save()
editor.annotation_input.wait_for_and_send_keys("source annotation")
self.assert_workflow_has_changes_and_save()
editor.configure_output(output="out_file1").wait_for_and_click()
output_label = editor.label_output(output="out_file1")
self.set_text_element(output_label, "workflow output label")
self.set_text_element(editor.rename_output, "renamed_output")
editor.change_datatype.wait_for_and_click()
editor.select_dataype_text_search.wait_for_and_send_keys("bam")
editor.select_datatype(datatype="bam").wait_for_and_click()
self.set_text_element(editor.add_tags, "#crazynewtag")
self.set_text_element(editor.remove_tags, "#oldboringtag")
self.sleep_for(self.wait_types.UX_RENDER)
cat_node.clone.wait_for_and_click()
editor.label_input.wait_for_and_send_keys("cloned label")
output_label = editor.label_output(output="out_file1")
self.set_text_element(output_label, "cloned output label")
self.sleep_for(self.wait_types.UX_RENDER)
self.assert_workflow_has_changes_and_save()
edited_workflow = self.workflow_populator.download_workflow(workflow_id)
source_step = next(iter(step for step in edited_workflow["steps"].values() if step["label"] == "source label"))
cloned_step = next(iter(step for step in edited_workflow["steps"].values() if step["label"] == "cloned label"))
assert source_step["annotation"] == cloned_step["annotation"] == "source annotation"
assert source_step["workflow_outputs"][0]["label"] == "workflow output label"
assert cloned_step["workflow_outputs"][0]["label"] == "cloned output label"
assert len(source_step["post_job_actions"]) == len(cloned_step["post_job_actions"]) == 4
assert source_step["post_job_actions"] == cloned_step["post_job_actions"]
[docs] @selenium_test
def test_editor_embed_workflow(self):
workflow_populator = self.workflow_populator
child_workflow_name = self._get_random_name()
workflow_populator.upload_yaml_workflow(WORKFLOW_OPTIONAL_TRUE_INPUT_COLLECTION, name=child_workflow_name)
parent_workflow_id = workflow_populator.upload_yaml_workflow(
"""class: GalaxyWorkflow
inputs: []
steps:
- tool_id: multiple_versions
tool_version: 0.1
label: multiple_versions
state:
foo: bar
"""
)
self.workflow_index_open()
self.workflow_index_click_option("Edit")
editor = self.components.workflow_editor
editor.canvas_body.wait_for_visible()
editor.tool_menu.wait_for_visible()
editor.tool_menu_section_link(section_name="workflows").wait_for_and_click()
editor.workflow_link(workflow_title=child_workflow_name).wait_for_and_click()
self.sleep_for(self.wait_types.UX_RENDER)
self.assert_workflow_has_changes_and_save()
workflow = self.workflow_populator.download_workflow(parent_workflow_id)
subworkflow_step = workflow["steps"]["1"]
assert subworkflow_step["name"] == child_workflow_name
assert subworkflow_step["type"] == "subworkflow"
assert subworkflow_step["subworkflow"]["a_galaxy_workflow"] == "true"
[docs] @selenium_test
def test_workflow_bookmarking(self):
@retry_during_transitions
def assert_workflow_bookmarked_status(target_status):
name_matches = [c.text == new_workflow_name for c in self.components.tool_panel.workflow_names.all()]
status = any(name_matches)
self.assertTrue(status == target_status)
new_workflow_name = self.workflow_create_new(clear_placeholder=True)
# Assert workflow not initially bookmarked.
assert_workflow_bookmarked_status(False)
self.components.workflow_editor.canvas_body.wait_for_visible()
self.wait_for_selector_absent_or_hidden(self.modal_body_selector())
self.components.masthead.workflow.wait_for_and_click()
# parse workflow table
table_elements = self.workflow_index_table_elements()
self.sleep_for(self.wait_types.UX_RENDER)
bookmark_td = table_elements[0].find_elements_by_tag_name("td")[4]
# get bookmark pseudo element
# https://stackoverflow.com/questions/45427223/click-on-pseudo-element-using-selenium
self.action_chains().move_to_element_with_offset(bookmark_td, 20, 20).click().perform()
self.sleep_for(self.wait_types.UX_TRANSITION)
# search for bookmark in tools menu
self.components.tool_panel.search.wait_for_and_send_keys(new_workflow_name)
assert_workflow_bookmarked_status(True)
[docs] 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)
[docs] 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(f"#{source_id}")
sink_element = self.driver.find_element_by_css_selector(f"#{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()
[docs] 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()
[docs] 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()
[docs] def open_in_workflow_editor(self, yaml_content, auto_layout=True):
name = self.workflow_upload_yaml_with_random_name(yaml_content)
self.workflow_index_open()
self.workflow_index_open_with_name(name)
if auto_layout:
self.workflow_editor_click_option("Auto Layout")
return name
[docs] 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
[docs] 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()
[docs] def workflow_index_open_with_name(self, name):
self.workflow_index_open()
self.workflow_index_search_for(name)
self.workflow_index_click_option("Edit")
[docs] 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
[docs] @retry_assertion_during_transitions
def assert_wf_name_is(self, expected_name):
edit_name_element = self.components.workflow_editor.edit_name.wait_for_visible()
actual_name = edit_name_element.get_attribute("value")
assert expected_name in actual_name, f"'{expected_name}' unequal name '{actual_name}'"
[docs] @retry_assertion_during_transitions
def assert_wf_annotation_is(self, expected_annotation):
edit_annotation = self.components.workflow_editor.edit_annotation
edit_annotation_element = edit_annotation.wait_for_visible()
actual_annotation = edit_annotation_element.get_attribute("value")
assert (
expected_annotation in actual_annotation
), f"'{expected_annotation}' unequal annotation '{actual_annotation}'"
[docs] @retry_assertion_during_transitions
def assert_modal_has_text(self, expected_text):
modal_element = self.components.workflow_editor.state_modal_body.wait_for_visible()
text = modal_element.text
assert expected_text in text, f"Failed to find expected text [{expected_text}] in modal text [{text}]"
[docs] def move_center_of_canvas(self, xoffset=0, yoffset=0):
canvas = self.driver.find_element_by_id("canvas-container")
chains = ActionChains(self.driver)
chains.click_and_hold(canvas).move_by_offset(xoffset=xoffset, yoffset=yoffset).release().perform()