test_workflow_editor.py

import json
from typing import (
    cast,
    TYPE_CHECKING,
)

if TYPE_CHECKING:
    from galaxy.selenium.has_playwright_driver import HasPlaywrightDriver

import yaml
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.remote.webelement import WebElement

from galaxy.selenium.navigates_galaxy import ColumnDefinition
from galaxy.selenium.web_element_protocol import WebElementProtocol
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,
    RunsWorkflows,
    selenium_only,
    selenium_test,
    SeleniumTestCase,
    UsesWorkflowAssertions,
)

CHIPSEQ_COLUMNS = [
    ColumnDefinition(
        "Condition",
        "The column is used to specify the specific experimental condition that each sample represents. There is no formal restriction on this column, but values should be kept short for readable reports.",
        "Text",
    ),
    ColumnDefinition(
        "Replicate",
        "This column is used to specify a replicate number for the experiment.",
        "Integer",
        optional=True,
        default_value="",
    ),
    ColumnDefinition(
        "Control",
        "If set, this should reference the element identifier corresponding to the control for this sample.",
        "Element Identifier",
        optional=True,
    ),
]


class TestWorkflowEditor(SeleniumTestCase, RunsWorkflows, UsesWorkflowAssertions):
    ensure_registered = True

    @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()

        # shouldn't have changes on fresh load
        save_button = self.components.workflow_editor.save_button
        save_button.assert_disabled()

        self.screenshot("workflow_editor_blank")

        self.hover_over(self.components._.left_panel_drag.wait_for_visible())

        self.workflow_editor_maximize_center_pane()

        self.screenshot("workflow_editor_center_pane_maximized")

    @selenium_test
    def test_edit_annotation(self):
        editor = self.components.workflow_editor
        annotation = "new_annotation_test"
        name = self.create_and_wait_for_new_workflow_in_editor(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)

    @selenium_test
    def test_edit_name(self):
        name = self.create_and_wait_for_new_workflow_in_editor()
        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)

    @selenium_test
    def test_edit_license(self):
        editor = self.components.workflow_editor
        name = self.create_and_wait_for_new_workflow_in_editor()
        editor.canvas_body.wait_for_visible()
        assert "Do not specify" in self.workflow_editor_license_text()

        self.workflow_editor_set_license("MIT")
        self.workflow_editor_click_save()

        self.workflow_index_open_with_name(name)
        assert "MIT" in self.workflow_editor_license_text()

    @selenium_test
    def test_parameter_regex_validation(self):
        editor = self.components.workflow_editor
        workflow_run = self.components.workflow_run

        parameter_name = "text_param"
        name = self.create_and_wait_for_new_workflow_in_editor()
        self.workflow_editor_add_input("parameter_input")
        editor.label_input.wait_for_and_send_keys(parameter_name)
        # this really should be parameterized with the repeat name
        self.components.tool_form.repeat_insert.wait_for_and_click()
        self.components.tool_form.parameter_input(
            parameter="parameter_definition|validators_0|regex_match"
        ).wait_for_and_send_keys("moocow.*")
        self.components.tool_form.parameter_input(
            parameter="parameter_definition|validators_0|regex_doc"
        ).wait_for_and_send_keys("input must start with moocow")
        self.save_after_node_form_changes()

        self.workflow_run_with_name(name)
        self.sleep_for(self.wait_types.UX_TRANSITION)
        input_element = workflow_run.simplified_input(label=parameter_name).wait_for_and_click()
        input_element.send_keys("startswrong")
        workflow_run.run_workflow_disabled.wait_for_absent()
        workflow_run.run_error.assert_absent_or_hidden()
        self.workflow_run_submit()
        element = workflow_run.run_error.wait_for_present()
        assert "input must start with moocow" in element.text

    @selenium_test
    def test_int_parameter_minimum_validation(self):
        editor = self.components.workflow_editor
        workflow_run = self.components.workflow_run

        parameter_name = "int_param"
        name = self.create_and_wait_for_new_workflow_in_editor()
        self.workflow_editor_add_input("parameter_input")
        editor.label_input.wait_for_and_send_keys(parameter_name)
        select_field = self.components.tool_form.parameter_select(parameter="parameter_definition|parameter_type")
        self.select_set_value(select_field, "integer")
        self.components.tool_form.parameter_input(parameter="parameter_definition|min").wait_for_and_send_keys("4")
        self.save_after_node_form_changes()

        self.workflow_run_with_name(name)
        self.sleep_for(self.wait_types.UX_TRANSITION)
        input_element = workflow_run.simplified_input(label=parameter_name).wait_for_and_click()
        input_element.send_keys("3")
        workflow_run.run_workflow_disabled.wait_for_absent()
        workflow_run.run_error.assert_absent_or_hidden()
        self.workflow_run_submit()
        element = workflow_run.run_error.wait_for_present()
        # follow up with a bigger PR to just make this (4 <= value) right? need to set default message
        # in parameter validators
        assert "Value ('3') must fulfill (4 <= value <= +infinity)" in element.text, element.text

    @selenium_test
    def test_float_parameter_maximum_validation(self):
        editor = self.components.workflow_editor
        workflow_run = self.components.workflow_run

        parameter_name = "float_param"
        name = self.create_and_wait_for_new_workflow_in_editor()
        self.workflow_editor_add_input("parameter_input_float")
        editor.label_input.wait_for_and_send_keys(parameter_name)
        self.components.tool_form.parameter_input(parameter="parameter_definition|max").wait_for_and_send_keys("3.14")
        self.save_after_node_form_changes()

        self.workflow_run_with_name(name)
        self.sleep_for(self.wait_types.UX_TRANSITION)
        input_element = workflow_run.simplified_input(label=parameter_name).wait_for_and_click()
        input_element.send_keys("3.2")
        workflow_run.run_workflow_disabled.wait_for_absent()
        workflow_run.run_error.assert_absent_or_hidden()
        self.workflow_run_submit()
        element = workflow_run.run_error.wait_for_present()
        # see message in test test_int_parameter_minimum_validation about making this a little more human
        # friendly.
        assert "Value ('3.2') must fulfill (-infinity <= value <= 3.14)" in element.text, element.text

    @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.components.workflows.edit_button.wait_for_and_click()
        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()
        self.components.tool_form.parameter_input(parameter="select_single").wait_for_and_send_keys("parameter value")
        self.save_after_node_form_changes()
        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"] == ""

    @selenium_test
    def test_data_input(self):
        editor = self.components.workflow_editor

        name = self.workflow_create_new()
        self.workflow_editor_add_input(item_name="data_input")
        self.screenshot("workflow_editor_data_input_new")
        self.workflow_editor_set_node_label("input1")
        self.workflow_editor_set_node_annotation("my cool annotation")
        self.sleep_for(self.wait_types.UX_RENDER)
        self.screenshot("workflow_editor_data_input_filled_in")
        self.workflow_editor_click_save()

        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
        annotation = editor.annotation_input.wait_for_value()
        assert annotation == "my cool annotation", 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()
        self.workflow_editor_add_input(item_name="data_collection_input")
        self.screenshot("workflow_editor_data_collection_input_new")
        editor.label_input.wait_for_and_send_keys("input1")
        editor.annotation_input.wait_for_and_send_keys("my cool annotation")
        self.sleep_for(self.wait_types.UX_RENDER)
        self.screenshot("workflow_editor_data_collection_input_filled_in")
        self.workflow_editor_click_save()

        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
        annotation = editor.annotation_input.wait_for_value()
        assert annotation == "my cool annotation", annotation
        data_input_node.destroy.wait_for_and_click()
        data_input_node.wait_for_absent()
        self.sleep_for(self.wait_types.UX_RENDER)
        self.screenshot("workflow_editor_data_collection_input_deleted")

    @selenium_only("Not yet migrated to support Playwright backend")
    @selenium_test
    def test_collection_input_sample_sheet_chipseq_example(self):
        editor = self.components.workflow_editor

        self.workflow_create_new()
        self.workflow_editor_add_input(item_name="data_collection_input")
        self.screenshot("workflow_editor_data_collection_sample_sheet_input_new")
        editor.label_input.wait_for_and_send_keys("input1")
        editor.annotation_input.wait_for_and_send_keys("chipseq example input")
        self.sleep_for(self.wait_types.UX_RENDER)
        editor.collection_type_input.wait_for_and_clear_and_send_keys("sample_sheet:paired")

        self.workflow_editor_enter_column_definitions(CHIPSEQ_COLUMNS)

        self.screenshot("workflow_editor_data_collection_sample_sheet_input_filled_in")
        self.workflow_editor_click_save()

        workflow = self._download_current_workflow()
        tool_state = json.loads(workflow["steps"]["0"]["tool_state"])
        assert tool_state["collection_type"] == "sample_sheet:paired"
        column_definitions = tool_state["column_definitions"]
        assert len(column_definitions) == len(CHIPSEQ_COLUMNS)
        condition = column_definitions[0]
        assert condition["name"] == "Condition"
        assert condition["description"] == CHIPSEQ_COLUMNS[0].description
        assert condition["type"] == "string"
        assert condition["optional"] is False

        replicate = column_definitions[1]
        assert replicate["name"] == "Replicate"
        assert replicate["type"] == "int"
        assert replicate["optional"] is True
        assert replicate["default_value"] is None

        control = column_definitions[2]
        assert control["name"] == "Control"
        assert control["type"] == "element_identifier"
        assert control["optional"] is True

    @selenium_test
    def test_data_column_input_editing(self):
        self.open_in_workflow_editor("""
class: GalaxyWorkflow
steps:
  column_param_list:
    tool_id: column_param_list
    state:
      col: ["1","2","3"]
      col_names: ["a", "b", "c"]
      """)
        editor = self.components.workflow_editor
        node = editor.node._(label="column_param_list")
        node.title.wait_for_and_click()
        columns = self.components.tool_form.parameter_textarea(parameter="col")
        textarea_columns = columns.wait_for_visible()
        assert textarea_columns.get_attribute("value") == "1\n2\n3"
        column_names = self.components.tool_form.parameter_textarea(parameter="col_names")
        textarea_column_names = column_names.wait_for_visible()
        assert textarea_column_names.get_attribute("value") == "a\nb\nc"
        self.sleep_for(self.wait_types.UX_RENDER)
        self.set_text_element(columns, "4\n5\n6")
        self.sleep_for(self.wait_types.UX_RENDER)
        self.assert_workflow_has_changes_and_save()
        self.refresh()
        node.title.wait_for_and_click()
        textarea_columns = columns.wait_for_visible()
        assert textarea_columns.get_attribute("value") == "4\n5\n6"

    @selenium_test
    def test_integer_input(self):
        editor = self.components.workflow_editor

        name = self.workflow_create_new(save_workflow=False)
        self.workflow_editor_add_input(item_name="parameter_input")
        self.screenshot("workflow_editor_parameter_input_new")
        editor.label_input.wait_for_and_send_keys("input1")
        editor.annotation_input.wait_for_and_send_keys("my cool annotation")
        self.sleep_for(self.wait_types.UX_RENDER)
        self.screenshot("workflow_editor_parameter_input_filled_in")
        self.workflow_editor_click_save()

        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
        annotation = editor.annotation_input.wait_for_value()
        assert annotation == "my cool annotation", annotation
        data_input_node.destroy.wait_for_and_click()
        data_input_node.wait_for_absent()
        self.sleep_for(self.wait_types.UX_RENDER)
        self.screenshot("workflow_editor_parameter_input_deleted")

    @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_node.wait_for_and_click()
        tool_input = tool_node.input_terminal(name="inttest")
        self.hover_over(tool_input.wait_for_visible())
        tool_node.connector_destroy_callout(name="inttest").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")

    @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")

    @selenium_test
    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")

    @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")
        self.hover_over(cat_input.wait_for_visible())
        cat_node.connector_destroy_callout(name="input1").wait_for_visible()
        self.screenshot("workflow_editor_connection_callout")
        cat_node.connector_destroy_callout(name="input1").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_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")

    @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_best_practices_input_label(self):
        editor = self.components.workflow_editor
        annotation = "best_practices_input_label"
        self.workflow_create_new(annotation=annotation)
        self.workflow_editor_add_input(item_name="data_input")
        editor.tool_bar.best_practices.wait_for_and_click()
        best_practices = editor.best_practices
        section_element = best_practices.section_input_metadata.wait_for_present()
        assert section_element.get_attribute("data-lint-status") == "warning"
        item = best_practices.item_input_metadata(index=0)
        element = item.wait_for_present()
        assert element.get_attribute("data-missing-label") == "true"
        assert element.get_attribute("data-missing-annotation") == "true"

        item.wait_for_and_click()
        self.workflow_editor_set_node_label("best practice input")

        # editor.tool_bar.best_practices.wait_for_and_click()
        element = item.wait_for_present()
        assert element.get_attribute("data-missing-label") == "false"
        assert element.get_attribute("data-missing-annotation") == "true"

        self.workflow_editor_set_node_annotation("informative annotation")

        @retry_assertion_during_transitions
        def assert_linting_input_metadata_okay():
            section_element = best_practices.section_input_metadata.wait_for_present()
            assert section_element.get_attribute("data-lint-status") == "ok"

        assert_linting_input_metadata_okay()

    @selenium_only("Not yet migrated to support Playwright backend")
    @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.components.workflow_editor.save_as_activity.wait_for_and_click()
        self.components.workflow_editor.save_as_modal_close.wait_for_and_click()

    @selenium_test
    def test_editor_tool_upgrade(self):
        workflow_populator = self.workflow_populator
        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
        """,
            exact_tools=True,
        )
        self.workflow_index_open()
        self.components.workflows.edit_button.wait_for_and_click()
        editor = self.components.workflow_editor
        self.workflow_editor_set_tool_vesrion("0.2", node="multiple_versions")
        self.screenshot("workflow_editor_version_update")
        self.sleep_for(self.wait_types.UX_RENDER)
        self.assert_workflow_has_changes_and_save()
        workflow = self.workflow_populator.download_workflow(workflow_id)
        assert workflow["steps"]["0"]["tool_version"] == "0.2"
        editor.node._(label="multiple_versions").wait_for_and_click()
        editor.tool_version_button.wait_for_and_click()
        assert self.select_dropdown_item("Switch to 0.1+galaxy6"), "Switch to tool version dropdown item not found"
        self.screenshot("workflow_editor_version_downgrade")
        self.sleep_for(self.wait_types.UX_RENDER)
        self.assert_workflow_has_changes_and_save()
        workflow = self.workflow_populator.download_workflow(workflow_id)
        assert workflow["steps"]["0"]["tool_version"] == "0.1+galaxy6"

    @selenium_test
    def test_editor_tool_upgrade_all_tools(self):
        editor = self.components.workflow_editor
        annotation = "upgarde_all_test"
        self.workflow_create_new(annotation=annotation)
        self.workflow_editor_add_tool_step("multiple_versions")
        self.workflow_editor_set_node_label(label="target label")
        self.workflow_editor_set_tool_vesrion("0.1")
        self.assert_workflow_has_changes_and_save()

        editor.tool_bar.upgrade_all.wait_for_and_click()
        self.workflow_editor_ensure_tool_form_open(node=0)
        node = self.components.tool_form.tool_version.wait_for_present()
        version = node.get_attribute("data-version")
        assert version == "0.2"

    @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.components.workflows.edit_button.wait_for_and_click()
        self.assert_modal_has_text("Using version '0.2' instead of version '0.0.1'")
        self.screenshot("workflow_editor_tool_upgrade")
        self.components.workflow_editor.state_upgrade_modal_close.wait_for_and_click()
        self.assert_workflow_has_changes_and_save()

    @selenium_test
    def test_editor_subworkflow_tool_upgrade_message(self):
        workflow_populator = self.workflow_populator
        embedded_workflow = yaml.safe_load(WORKFLOW_WITH_OLD_TOOL_VERSION)
        # Create invalid tool state
        embedded_workflow["steps"]["mul_versions"]["state"]["inttest"] = "Invalid"
        outer_workflow = yaml.safe_load("""
class: GalaxyWorkflow
inputs:
  outer_input: data
steps:
  nested_workflow:
    run: {}
    in:
      input1: outer_input
        """)
        outer_workflow["steps"]["nested_workflow"]["run"] = embedded_workflow
        workflow_populator.upload_yaml_workflow(json.dumps(outer_workflow), exact_tools=True)
        self.workflow_index_open()
        self.components.workflows.edit_button.wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)
        self.assert_modal_has_text("Using version '0.2' instead of version '0.0.1'")
        self.assert_modal_has_text("Parameter 'inttest': an integer or workflow parameter is required")
        self.screenshot("workflow_editor_subworkflow_tool_upgrade")
        self.workflow_editor_dismiss_state_upgrade_modal()
        self.assert_workflow_has_changes_and_save()
        # Ensure that the state upgrade modal is hidden by the end, to ensure AXE checks pass (no dialog should be present)
        self.workflow_editor_dismiss_state_upgrade_modal()

    @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)

    @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_datatype_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_connection_invalid("create_2#out_file1", "checksum#input")
        # Assert save button
        save_button = self.components.workflow_editor.save_button
        save_button.wait_for_and_click()
        # Will trigger confirmation modal
        self.components.confirm_dialog.ok_button.wait_for_and_click()
        self.components.confirm_dialog._.wait_for_absent_or_hidden()
        # Make connection valid again
        editor.configure_output(output="out_file1").wait_for_and_click()
        editor.change_datatype.wait_for_and_click()
        editor.select_datatype_text_search.wait_for_and_send_keys("tabular")
        editor.select_datatype(datatype="tabular").wait_for_and_click()
        # Assert connection is valid
        self.assert_connected("create_2#out_file1", "checksum#input")

    @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
    out:
      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")

    @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
            out:
              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")

    @selenium_test
    def test_edit_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
""")
        # Save after auto-layout so there are no unsaved changes
        self.assert_workflow_has_changes_and_save()
        editor = self.components.workflow_editor
        node = editor.node._(label="nested_workflow")
        node.wait_for_and_click()
        editor.edit_subworkflow.wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)
        node = editor.node._(label="create_2")
        node.wait_for_and_click()

    @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.components.workflows.edit_button.wait_for_and_click()
        editor = self.components.workflow_editor
        self.workflow_editor_set_node_label(label="source label", node="first_cat")
        # 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_datatype_text_search.wait_for_and_send_keys("bam")
        editor.select_datatype(datatype="bam").wait_for_and_click()
        editor.add_tags_button.wait_for_and_click()
        editor.add_tags_input.wait_for_and_send_keys("#crazynewtag" + Keys.ENTER + Keys.ESCAPE)
        editor.remove_tags_button.wait_for_and_click()
        editor.remove_tags_input.wait_for_and_send_keys("#oldboringtag" + Keys.ENTER + Keys.ESCAPE)
        self.sleep_for(self.wait_types.UX_RENDER)
        cat_node.clone.wait_for_and_click()
        cloned_node = editor.node.by_id(id=2)
        cloned_node.wait_for_and_click()
        editor.label_input.wait_for_and_send_keys(Keys.BACKSPACE * 20)
        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"]

    @selenium_test
    def test_editor_embed_workflow(self):
        self.setup_subworkflow()

    def setup_subworkflow(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:
  input_collection:
    type: collection
    collection_type: "list"
steps:
  - tool_id: multiple_versions
    tool_version: "0.1"
    label: multiple_versions
    state:
      foo: bar
        """)
        self.workflow_index_open()
        self.components.workflows.edit_button.wait_for_and_click()
        editor = self.components.workflow_editor
        editor.canvas_body.wait_for_visible()
        self.workflow_editor_add_subworkflow(child_workflow_name)
        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"]["2"]
        assert subworkflow_step["name"] == child_workflow_name
        assert subworkflow_step["type"] == "subworkflow"
        assert subworkflow_step["subworkflow"]["a_galaxy_workflow"] == "true"
        self.workflow_editor_connect("input_collection#output", f"{child_workflow_name}#input1")
        self.assert_connected("input_collection#output", f"{child_workflow_name}#input1")
        self.assert_workflow_has_changes_and_save()
        workflow = self.workflow_populator.download_workflow(parent_workflow_id)
        subworkflow_step = workflow["steps"]["2"]
        assert subworkflow_step["input_connections"]["input1"]["input_subworkflow_step_id"] == 0
        return child_workflow_name

    @selenium_test
    def test_editor_insert_steps(self):
        steps_to_insert = self.workflow_upload_yaml_with_random_name(WORKFLOW_SIMPLE_CAT_TWICE)
        annotation = "insert step test"
        self.workflow_create_new(annotation=annotation)
        self.workflow_editor_add_input(item_name="data_input")
        editor = self.components.workflow_editor
        editor.canvas_body.wait_for_visible()
        self.workflow_editor_add_steps(steps_to_insert)
        self.assert_connected("input1#output", "first_cat#input1")
        self.assert_workflow_has_changes_and_save()
        workflow = self._download_current_workflow()
        assert len(workflow["steps"]) == 3

    def _download_current_workflow(self):
        self.sleep_for(self.wait_types.DATABASE_OPERATION)
        workflow_id = self.current_url.split("id=")[1]
        workflow = self.workflow_populator.download_workflow(workflow_id)
        return workflow

    def _pick_value_select_mode(self, label):
        mode_selector = "div.ui-form-element[id='form-element-mode']"
        container = self.wait_for_selector(mode_selector)
        trigger = container.find_element(By.CSS_SELECTOR, ".multiselect__select")
        trigger.click()
        self.sleep_for(self.wait_types.UX_RENDER)
        js = """
            var label = arguments[0];
            var container = document.querySelector('#form-element-mode');
            var items = container.querySelectorAll('.multiselect__element');
            for (var i = 0; i < items.length; i++) {
                if (items[i].textContent.trim() === label) {
                    items[i].querySelector('.multiselect__option').click();
                    return true;
                }
            }
            return false;
        """
        result = self.execute_script(js, label)
        assert result, f"Mode option '{label}' not found"

    @selenium_test
    def test_pick_value_add_from_palette(self):
        self.workflow_create_new(annotation="pick value test")
        self.workflow_editor_add_input(item_name="pick_value")
        editor = self.components.workflow_editor
        editor.node._(label="Pick Value").wait_for_present()

    @selenium_test
    def test_pick_value_mode_selection(self):
        self.workflow_create_new(annotation="pick value mode test")
        self.workflow_editor_add_input(item_name="pick_value")
        editor = self.components.workflow_editor
        node = editor.node._(label="Pick Value")
        node.wait_for_and_click()
        self._pick_value_select_mode("All non-null (as collection)")
        self.sleep_for(self.wait_types.UX_RENDER)
        self.assert_workflow_has_changes_and_save()
        workflow = self._download_current_workflow()
        pick_step = [s for s in workflow["steps"].values() if s["type"] == "pick_value"][0]
        tool_state = json.loads(pick_step["tool_state"])
        assert tool_state["mode"] == "all_non_null"

    @selenium_test
    def test_pick_value_terminals(self):
        self.workflow_create_new(annotation="pick value terminals test")
        self.workflow_editor_add_input(item_name="pick_value")
        editor = self.components.workflow_editor
        node = editor.node._(label="Pick Value")
        node.input_terminal(name="input_0").wait_for_present()
        node.input_terminal(name="input_1").wait_for_present()
        node.output_terminal(name="output").wait_for_present()

    @selenium_test
    def test_pick_value_connect_inputs(self):
        self.workflow_create_new(annotation="pick value connections test")
        self.workflow_editor_add_input(item_name="data_input")
        editor = self.components.workflow_editor
        editor.label_input.wait_for_and_send_keys("input_data")
        self.tool_open("cat1")
        self.sleep_for(self.wait_types.UX_RENDER)
        editor.label_input.wait_for_and_send_keys("branch_a")
        self.workflow_editor_add_input(item_name="pick_value")
        editor.label_input.wait_for_and_send_keys("pick")
        self.components.workflow_editor.tool_bar.auto_layout.wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)
        self.workflow_editor_connect("input_data#output", "branch_a#input1")
        self.workflow_editor_connect("branch_a#out_file1", "pick#input_0")
        self.assert_connected("branch_a#out_file1", "pick#input_0")

    @selenium_test
    def test_pick_value_grow_on_connect(self):
        self.open_in_workflow_editor("""
class: GalaxyWorkflow
inputs:
  input_data: data
steps:
  branch_a:
    tool_id: cat
    in:
      input1: input_data
  branch_b:
    tool_id: cat
    in:
      input1: input_data
  pick:
    type: pick_value
    state:
      mode: first_non_null
    in:
      input_0: branch_a/out_file1
      input_1: branch_b/out_file1
""")
        editor = self.components.workflow_editor
        pick_node = editor.node._(label="pick")
        pick_node.input_terminal(name="input_0").wait_for_present()
        pick_node.input_terminal(name="input_1").wait_for_present()
        # With 2 connections, grow-on-connect should have created a 3rd empty terminal
        pick_node.input_terminal(name="input_2").wait_for_present()

    @selenium_test
    def test_pick_value_conditional_workflow_roundtrip(self):
        self.open_in_workflow_editor("""
class: GalaxyWorkflow
inputs:
  input_data: data
steps:
  branch_a:
    tool_id: cat
    in:
      input1: input_data
    when: $(true)
  branch_b:
    tool_id: cat
    in:
      input1: input_data
    when: $(false)
  pick:
    type: pick_value
    state:
      mode: first_non_null
    in:
      input_0: branch_a/out_file1
      input_1: branch_b/out_file1
""")
        editor = self.components.workflow_editor
        pick_node = editor.node._(label="pick")
        pick_node.wait_for_present()
        self.assert_connected("branch_a#out_file1", "pick#input_0")
        self.assert_connected("branch_b#out_file1", "pick#input_1")
        pick_node.output_terminal(name="output").wait_for_present()
        workflow = self._download_current_workflow()
        pick_step = [s for s in workflow["steps"].values() if s["type"] == "pick_value"][0]
        tool_state = json.loads(pick_step["tool_state"])
        assert tool_state["mode"] == "first_non_null"
        assert len(pick_step["input_connections"]) == 2

    @selenium_test
    def test_pick_value_output_type_changes_with_mode(self):
        self.open_in_workflow_editor("""
class: GalaxyWorkflow
inputs:
  input_data: data
steps:
  branch_a:
    tool_id: cat
    in:
      input1: input_data
  pick:
    type: pick_value
    state:
      mode: first_non_null
    in:
      input_0: branch_a/out_file1
""")
        editor = self.components.workflow_editor
        pick_node = editor.node._(label="pick")
        pick_node.wait_for_and_click()
        self._pick_value_select_mode("All non-null (as collection)")
        self.sleep_for(self.wait_types.UX_RENDER)
        self.assert_workflow_has_changes_and_save()
        workflow = self._download_current_workflow()
        pick_step = [s for s in workflow["steps"].values() if s["type"] == "pick_value"][0]
        tool_state = json.loads(pick_step["tool_state"])
        assert tool_state["mode"] == "all_non_null"

    @selenium_test
    def test_pick_value_change_datatype_pja(self):
        self.open_in_workflow_editor("""
class: GalaxyWorkflow
inputs:
  input_data: data
steps:
  branch_a:
    tool_id: cat
    in:
      input1: input_data
  pick:
    type: pick_value
    state:
      mode: first_non_null
    in:
      input_0: branch_a/out_file1
""")
        editor = self.components.workflow_editor
        pick_node = editor.node._(label="pick")
        pick_node.wait_for_and_click()
        # Expand the output card — PJA controls are inside a collapsed FormCard
        editor.configure_output(output="output").wait_for_and_click()
        editor.change_datatype.wait_for_and_click()
        editor.select_datatype_text_search.wait_for_and_send_keys("bam")
        editor.select_datatype(datatype="bam").wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)
        self.assert_workflow_has_changes_and_save()
        workflow = self._download_current_workflow()
        pick_step = [s for s in workflow["steps"].values() if s["type"] == "pick_value"][0]
        pjas = pick_step["post_job_actions"]
        assert "ChangeDatatypeActionoutput" in pjas
        assert pjas["ChangeDatatypeActionoutput"]["action_arguments"]["newtype"] == "bam"

    @selenium_test
    def test_pick_value_rename_pja(self):
        self.open_in_workflow_editor("""
class: GalaxyWorkflow
inputs:
  input_data: data
steps:
  branch_a:
    tool_id: cat
    in:
      input1: input_data
  pick:
    type: pick_value
    state:
      mode: first_non_null
    in:
      input_0: branch_a/out_file1
""")
        editor = self.components.workflow_editor
        pick_node = editor.node._(label="pick")
        pick_node.wait_for_and_click()
        editor.configure_output(output="output").wait_for_and_click()
        editor.rename_output.wait_for_and_clear_and_send_keys("my_picked_output")
        self.sleep_for(self.wait_types.UX_RENDER)
        self.assert_workflow_has_changes_and_save()
        workflow = self._download_current_workflow()
        pick_step = [s for s in workflow["steps"].values() if s["type"] == "pick_value"][0]
        pjas = pick_step["post_job_actions"]
        assert "RenameDatasetActionoutput" in pjas
        assert pjas["RenameDatasetActionoutput"]["action_arguments"]["newname"] == "my_picked_output"

    @selenium_test
    def test_pick_value_add_tags_pja(self):
        self.open_in_workflow_editor("""
class: GalaxyWorkflow
inputs:
  input_data: data
steps:
  branch_a:
    tool_id: cat
    in:
      input1: input_data
  pick:
    type: pick_value
    state:
      mode: first_non_null
    in:
      input_0: branch_a/out_file1
""")
        editor = self.components.workflow_editor
        pick_node = editor.node._(label="pick")
        pick_node.wait_for_and_click()
        editor.configure_output(output="output").wait_for_and_click()
        editor.add_tags_button.wait_for_and_click()
        tag_input = editor.add_tags_input.wait_for_visible()
        tag_input.send_keys("#picktag")
        self.send_enter(tag_input)
        self.send_escape(tag_input)
        self.sleep_for(self.wait_types.UX_RENDER)
        self.assert_workflow_has_changes_and_save()
        workflow = self._download_current_workflow()
        pick_step = [s for s in workflow["steps"].values() if s["type"] == "pick_value"][0]
        pjas = pick_step["post_job_actions"]
        assert "TagDatasetActionoutput" in pjas
        assert "picktag" in pjas["TagDatasetActionoutput"]["action_arguments"]["tags"]

    @selenium_test
    def test_pick_value_hides_actions_needing_a_job(self):
        self.open_in_workflow_editor("""
class: GalaxyWorkflow
inputs:
  input_data: data
steps:
  branch_a:
    tool_id: cat
    in:
      input1: input_data
  pick:
    type: pick_value
    state:
      mode: first_non_null
    in:
      input_0: branch_a/out_file1
""")
        editor = self.components.workflow_editor
        # A pick_value step has no job for these to act on, so they are not offered.
        # Select it first - the panel a selected node opens covers the nodes to its right.
        editor.node._(label="pick").wait_for_and_click()
        # The rest of the section still renders, otherwise the assertions below would
        # pass on a form that simply had not loaded.
        editor.configure_output(output="output").wait_for_visible()
        editor.email_notification.assert_absent()
        editor.output_cleanup.assert_absent()
        # A tool step does run a job, so it still gets both.
        editor.node._(label="branch_a").wait_for_and_click()
        editor.email_notification.wait_for_visible()
        editor.output_cleanup.wait_for_visible()

    @selenium_test
    def test_pick_value_compact_on_disconnect(self):
        self.open_in_workflow_editor("""
class: GalaxyWorkflow
inputs:
  input_data: data
steps:
  branch_a:
    tool_id: cat
    in:
      input1: input_data
  branch_b:
    tool_id: cat
    in:
      input1: input_data
  branch_c:
    tool_id: cat
    in:
      input1: input_data
  pick:
    type: pick_value
    state:
      mode: first_non_null
    in:
      input_0: branch_a/out_file1
      input_1: branch_b/out_file1
      input_2: branch_c/out_file1
""")
        editor = self.components.workflow_editor
        pick_node = editor.node._(label="pick")
        # Verify initial state: 3 connected + 1 empty from grow-on-connect
        self.assert_connected("branch_a#out_file1", "pick#input_0")
        self.assert_connected("branch_b#out_file1", "pick#input_1")
        self.assert_connected("branch_c#out_file1", "pick#input_2")
        pick_node.input_terminal(name="input_3").wait_for_present()
        # Disconnect middle input
        self.workflow_editor_destroy_connection("pick#input_1")
        self.sleep_for(self.wait_types.UX_RENDER)
        # branch_b should no longer be connected
        self.assert_not_connected("branch_b#out_file1", "pick#input_1")
        # branch_c should have been compacted from input_2 to input_1
        self.assert_connected("branch_a#out_file1", "pick#input_0")
        self.assert_connected("branch_c#out_file1", "pick#input_1")
        # Click pick node to mount FormPickValue watcher (triggers num_inputs shrink)
        pick_node.wait_for_and_click()
        pick_node.input_terminal(name="input_3").wait_for_absent_or_hidden()
        # Save and verify connections and num_inputs were compacted
        self.assert_workflow_has_changes_and_save()
        workflow = self._download_current_workflow()
        pick_step = [s for s in workflow["steps"].values() if s["type"] == "pick_value"][0]
        tool_state = json.loads(pick_step["tool_state"])
        assert tool_state["num_inputs"] == 2
        assert len(pick_step["input_connections"]) == 2
        assert "input_0" in pick_step["input_connections"]
        assert "input_1" in pick_step["input_connections"]

    @selenium_test
    def test_editor_create_conditional_step(self):
        editor = self.components.workflow_editor
        self.workflow_create_new(annotation="simple when step definition")
        # Insert a boolean parameter
        self.workflow_editor_add_input(item_name="parameter_input")
        param_type_element = editor.param_type_form.wait_for_present()
        self.switch_param_type(param_type_element, "Boolean")
        editor.label_input.wait_for_and_send_keys("param_input")
        # Insert cat tool
        self.tool_open("cat")
        self.sleep_for(self.wait_types.UX_RENDER)
        editor.label_input.wait_for_and_send_keys("downstream_step")
        # Insert head tool
        self.tool_open("head")

        self.components.workflow_editor.tool_bar.auto_layout.wait_for_and_click()

        self.sleep_for(self.wait_types.UX_RENDER)
        editor.label_input.wait_for_and_send_keys("conditional_step")
        # Connect head to cat
        self.workflow_editor_connect("conditional_step#out_file1", "downstream_step#input1")
        self.assert_connected("conditional_step#out_file1", "downstream_step#input1")
        # Make head tool conditional
        conditional_node = editor.node._(label="conditional_step")
        conditional_node.input_terminal(name="input").wait_for_present()
        # Assert no when input before making step conditional
        conditional_node.input_terminal(name="when").wait_for_absent()
        conditional_toggle = editor.step_when.wait_for_present()
        self.move_to_and_click(conditional_toggle)
        # Toggling conditional should cause when input to appear
        conditional_node.input_terminal(name="when").wait_for_present()
        self.move_to_and_click(conditional_toggle)
        # Toggling conditional should cause when input to disappear
        conditional_node.input_terminal(name="when").wait_for_absent()
        self.move_to_and_click(conditional_toggle)
        conditional_node.input_terminal(name="when").wait_for_present()
        # Output connection should be invalid, as output from conditional step is potentially null
        self.assert_connection_invalid("conditional_step#out_file1", "downstream_step#input1")
        downstream_step = editor.node._(label="downstream_step")
        downstream_step.destroy.wait_for_and_click()
        downstream_step.wait_for_absent()
        # Connect boolean input to when
        self.workflow_editor_connect("param_input#output", "conditional_step#when")
        self.assert_connected("param_input#output", "conditional_step#when")
        # Change boolean input parameter to invalid parameter type
        editor.node._(label="param_input").wait_for_and_click()
        param_type_element = editor.param_type_form.wait_for_present()
        self.switch_param_type(param_type_element, "Text")
        self.assert_connection_invalid("param_input#output", "conditional_step#when")
        editor.node_inspector_close.wait_for_and_click()
        self.workflow_editor_destroy_connection("conditional_step#when")
        # Make sure the when input is still shown
        conditional_node.input_terminal(name="when").wait_for_present()
        # Assert save button is disabled because of disconnected when
        save_button = self.components.workflow_editor.save_button
        save_button.wait_for_visible()
        # TODO: hook up best practice panel, disable save when "when" not connected
        # assert save_button.has_class("disabled")

    @selenium_only("Not yet migrated to support Playwright backend")
    @selenium_test
    def test_conditional_subworkflow_step(self):
        child_workflow_name = self.setup_subworkflow()
        editor = self.components.workflow_editor
        # Insert a boolean parameter
        self.workflow_editor_add_input(item_name="parameter_input")
        param_type_element = editor.param_type_form.wait_for_present()
        self.switch_param_type(param_type_element, "Boolean")
        editor.label_input.wait_for_and_send_keys("param_input")
        self.components.workflow_editor.tool_bar.auto_layout.wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)
        conditional_node = editor.node._(label=child_workflow_name)
        conditional_node.wait_for_and_click()
        conditional_toggle = editor.step_when.wait_for_present()
        self.action_chains().move_to_element(conditional_toggle).click().perform()
        conditional_node.input_terminal(name="when").wait_for_present()
        self.workflow_editor_connect("param_input#output", f"{child_workflow_name}#when")
        self.assert_connected("param_input#output", f"{child_workflow_name}#when")
        self.assert_workflow_has_changes_and_save()

    def switch_param_type(self, element, param_type):
        if self.backend_type == "playwright":
            pw_driver = cast("HasPlaywrightDriver", self._driver_impl)
            element.click()
            self.sleep_for(self.wait_types.UX_RENDER)
            pw_driver.page.keyboard.type(param_type)
            self.sleep_for(self.wait_types.UX_RENDER)
            pw_driver.page.keyboard.press("Enter")
        else:
            self.action_chains().move_to_element(element).click().pause(1).send_keys(param_type).pause(1).send_keys(
                Keys.ENTER
            ).perform()

    @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.components.workflows.edit_button.wait_for_and_click()
        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")
        self.workflow_editor_dismiss_state_upgrade_modal()

    @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.components.workflows.edit_button.wait_for_and_click()
        self.assert_modal_has_text("Tool is not installed")
        self.screenshot("workflow_editor_missing_tool")
        self.workflow_editor_dismiss_state_upgrade_modal()

    def tab_to(self, accessible_name, direction="forward"):
        for _ in range(100):
            ac = self.action_chains()
            if direction == "backwards":
                ac.key_down(Keys.SHIFT)
            ac.send_keys(Keys.TAB)
            if direction == "backwards":
                ac.key_down(Keys.SHIFT)
            ac.perform()
            if accessible_name in self.driver.switch_to.active_element.accessible_name:
                return self.driver.switch_to.active_element
        else:
            raise Exception(f"Could not tab to element containing '{accessible_name}' in aria-label")

    @selenium_only("Not yet migrated to support Playwright backend")
    @selenium_test
    def test_aria_connections_menu(self):
        self.open_in_workflow_editor(
            """
class: GalaxyWorkflow
inputs:
  input1: data
steps:
  first_cat:
    position:
      # Step should be positioned off-screen, tabbing should scroll to node
      top: 2000
      left: 2000
    tool_id: cat
    in:
      input1: input1
      queries_0|input2: input1
""",
            auto_layout=False,
        )
        self.assert_connected("input1#output", "first_cat#input1")
        self.screenshot("workflow_editor_connection_simple")
        self.components.workflow_editor.canvas_body.wait_for_and_click()
        output_connector = self.tab_to("Press space to see a list of available inputs")
        output_connector.send_keys(Keys.SPACE)
        assert self.driver.switch_to.active_element.text == "Disconnect from input1 in step 2: first_cat"
        self.driver.switch_to.active_element.send_keys(Keys.ENTER)
        self.assert_not_connected("input1#output", "first_cat#input1")
        self.action_chains().move_to_element(self.components.workflow_editor.canvas_body.wait_for_and_click()).perform()
        output_connector = self.tab_to("Press space to see a list of available inputs")
        output_connector.send_keys(Keys.SPACE)
        assert self.driver.switch_to.active_element.text == "Connect to input1 in step 2: first_cat"
        self.driver.switch_to.active_element.send_keys(Keys.ENTER)
        self.assert_connected("input1#output", "first_cat#input1")
        self.action_chains().move_to_element(self.components.workflow_editor.canvas_body.wait_for_and_click()).perform()
        output_connector = self.tab_to("Press space to see a list of available inputs")
        output_connector = self.tab_to("Press space to see a list of available inputs")
        output_connector.send_keys(Keys.SPACE)
        assert self.driver.switch_to.active_element.text == "No compatible input found in workflow"

    @selenium_test
    def test_insert_input_handling(self):
        self.open_in_workflow_editor("""class: GalaxyWorkflow
inputs: []
steps:
  build_list:
    tool_id: __BUILD_LIST__
        """)
        editor = self.components.workflow_editor
        node = editor.node._(label="build_list")
        node.wait_for_and_click()
        assert not node.has_class("input-terminal")
        self.components.tool_form.repeat_insert.wait_for_and_click()
        node.input_terminal(name="datasets_0|input").wait_for_present()
        self.components.tool_form.repeat_insert.wait_for_and_click()
        node.input_terminal(name="datasets_1|input").wait_for_present()
        self.assert_workflow_has_changes_and_save()

    @selenium_test
    def test_workflow_output_handling(self):
        self.open_in_workflow_editor(
            """
class: GalaxyWorkflow
inputs: []
outputs:
  first_out:
    outputSource: first/out_file1
  second_out:
    outputSource: second/out_file1
steps:
  first:
    tool_id: create_2
  second:
    tool_id: create_2
""",
            auto_layout=True,
        )
        editor = self.components.workflow_editor
        # assert both steps have one workflow output each
        first = editor.node._(label="first")
        first.workflow_output_toggle_active(name="out_file1").wait_for_visible()
        first.workflow_output_toggle_active(name="out_file2").wait_for_absent()
        second = editor.node._(label="second")
        second.workflow_output_toggle_active(name="out_file1").wait_for_visible()
        second.workflow_output_toggle_active(name="out_file2").wait_for_absent()
        # toggle out_file1 on second step, both outputs not active
        second.workflow_output_toggle(name="out_file1").wait_for_and_click()
        # just toggling outputs doesn't set a label
        second.workflow_output_toggle_active(name="out_file1").wait_for_absent()
        second.workflow_output_toggle_active(name="out_file2").wait_for_absent()
        # switch to first node
        editor.node._(label="first").wait_for_and_click()
        # toggle out_file1 on first step
        first.workflow_output_toggle(name="out_file1").wait_for_and_click()
        first.workflow_output_toggle_active(name="out_file1").wait_for_absent()
        # turn out_file1 back on
        first.workflow_output_toggle(name="out_file1").wait_for_and_click()
        first.workflow_output_toggle_active(name="out_file1").wait_for_visible()
        # turn out_file1 off
        first.workflow_output_toggle(name="out_file1").wait_for_and_click()
        first.workflow_output_toggle_active(name="out_file1").wait_for_absent()
        editor.node._(label="second").wait_for_and_click()
        editor.node._(label="first").wait_for_and_click()
        # make sure workflow outputs are both off for second step
        first.workflow_output_toggle_active(name="out_file1").wait_for_absent()
        first.workflow_output_toggle_active(name="out_file2").wait_for_absent()
        # add an output label
        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")
        # should indicate active workflow output
        first.workflow_output_toggle_active(name="out_file1").wait_for_visible()
        self.set_text_element(output_label, "")
        # deleting label also deletes active output
        first.workflow_output_toggle_active(name="out_file1").wait_for_absent()
        # set duplicate label
        output_label = editor.label_output(output="out_file1")
        self.set_text_element(output_label, "workflow output label")
        editor.node._(label="second").wait_for_and_click()
        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")
        # should show error
        editor.duplicate_label_error(output="out_file1").wait_for_visible()
        # make label unique
        self.set_text_element(output_label, "workflow output label2")
        # should not show error
        editor.duplicate_label_error(output="out_file1").wait_for_absent()

    @selenium_test
    def test_map_over_output_indicator(self):
        self.open_in_workflow_editor("""
class: GalaxyWorkflow
inputs:
  list:
    type: collection
    collection_type: "list"
  nested_list:
    type: collection
    collection_type: "list:list"
steps:
  filter:
    tool_id: __FILTER_FROM_FILE__
""")
        self.assert_node_output_is("filter#output_filtered", "any")
        self.workflow_editor_connect("list#output", "filter#input")
        self.assert_node_output_is("filter#output_filtered", "list")
        self.workflow_editor_connect("nested_list#output", "filter#how|filter_source")
        self.assert_node_output_is("filter#output_filtered", "list", "list:list:list")
        self.workflow_editor_destroy_connection("filter#how|filter_source")
        self.assert_node_output_is("filter#output_filtered", "list")

    @selenium_test
    def test_editor_place_comments(self):
        editor = self.components.workflow_editor

        self.workflow_create_new(annotation="simple workflow")
        self.sleep_for(self.wait_types.UX_RENDER)

        canvas = editor.canvas_body.wait_for_visible()

        # select text comment tool use all options and set font size to 2
        editor.tool_bar.tool(tool="text_comment").wait_for_and_click()
        editor.tool_bar.toggle_bold.wait_for_and_click()
        editor.tool_bar.toggle_italic.wait_for_and_click()
        editor.tool_bar.color(color="pink").wait_for_and_click()
        editor.tool_bar.font_size.wait_for_and_click()
        self.send_keys_to_page(Keys.LEFT * 5 + Keys.RIGHT)

        # place text comment
        self.mouse_drag(from_element=canvas, from_offset=(-200, -200), to_offset=(400, 110))

        self.send_keys_to_page("Hello World")

        # check if all options were applied
        comment_content: WebElementProtocol = editor.comment.text_inner.wait_for_visible()
        assert comment_content.text == "Hello World"
        comment_content_class = comment_content.get_attribute("class")
        assert comment_content_class
        assert "bold" in comment_content_class
        assert "italic" in comment_content_class

        # check for correct size
        width, height = self.get_element_size(editor.comment._.wait_for_visible())

        assert width == 400
        assert height == 110

        editor.comment.text_comment.wait_for_and_click()
        editor.comment.delete.wait_for_and_click()
        editor.comment.text_comment.wait_for_absent()

        # place and test markdown comment
        editor.tool_bar.tool(tool="markdown_comment").wait_for_and_click()
        editor.tool_bar.color(color="lime").wait_for_and_click()
        self.mouse_drag(from_element=canvas, from_offset=(-100, -100), to_offset=(200, 220))
        self.send_keys_to_page("# Hello World")

        editor.tool_bar.tool(tool="pointer").wait_for_and_click()

        markdown_comment_content: WebElementProtocol = editor.comment.markdown_rendered.wait_for_visible()
        assert markdown_comment_content.text == "Hello World"
        assert markdown_comment_content.find_element(By.TAG_NAME, "h2") is not None

        width, height = self.get_element_size(editor.comment._.wait_for_visible())

        assert width == 200
        assert height == 220

        editor.comment.markdown_rendered.wait_for_and_click()
        editor.comment.delete.wait_for_and_click()
        editor.comment.markdown_comment.wait_for_absent()

        # place and test frame comment
        editor.tool_bar.tool(tool="frame_comment").wait_for_and_click()
        editor.tool_bar.color(color="blue").wait_for_and_click()
        self.mouse_drag(from_element=canvas, from_offset=(-200, -150), to_offset=(400, 300))
        self.send_keys_to_page("My Frame")

        title: WebElementProtocol = editor.comment.frame_title.wait_for_visible()
        assert title.text == "My Frame"

        width, height = self.get_element_size(editor.comment._.wait_for_visible())

        assert width == 400
        assert height == 300

        editor.comment.frame_comment.wait_for_and_click()
        editor.comment.delete.wait_for_and_click()
        editor.comment.frame_comment.wait_for_absent()

        # test freehand and eraser
        editor.tool_bar.tool(tool="freehand_pen").wait_for_and_click()
        editor.tool_bar.color(color="green").wait_for_and_click()
        editor.tool_bar.line_thickness.wait_for_and_click()
        self.send_keys_to_page(Keys.RIGHT * 20)

        editor.tool_bar.smoothing.wait_for_and_click()
        self.send_keys_to_page(Keys.RIGHT * 10)

        self.mouse_drag(from_element=canvas, from_offset=(-100, -100), to_offset=(200, 200))

        editor.comment.freehand_comment.wait_for_visible()

        editor.tool_bar.color(color="black").wait_for_and_click()
        editor.tool_bar.line_thickness.wait_for_and_click()
        self.send_keys_to_page(Keys.LEFT * 20)
        self.mouse_drag(from_element=canvas, from_offset=(-100, -100), via_offsets=[(100, 200)], to_offset=(-200, 30))

        # test bulk remove freehand
        editor.tool_bar.remove_freehand.wait_for_and_click()
        editor.comment.freehand_comment.wait_for_absent()

        # place another freehand comment and test eraser
        editor.tool_bar.line_thickness.wait_for_and_click()
        self.send_keys_to_page(Keys.RIGHT * 20)
        editor.tool_bar.color(color="orange").wait_for_and_click()

        self.mouse_drag(from_element=canvas, from_offset=(-100, -100), to_offset=(200, 200))

        freehand_comment_a: WebElementProtocol = editor.comment.freehand_comment.wait_for_visible()

        # delete by clicking
        editor.tool_bar.tool(tool="freehand_eraser").wait_for_and_click()
        self.move_to_and_click(freehand_comment_a)

        editor.comment.freehand_comment.wait_for_absent()

        # delete by dragging
        editor.tool_bar.tool(tool="freehand_pen").wait_for_and_click()
        editor.tool_bar.color(color="yellow").wait_for_and_click()

        self.mouse_drag(from_element=canvas, from_offset=(-100, -100), to_offset=(200, 200))

        freehand_comment_b: WebElementProtocol = editor.comment.freehand_comment.wait_for_visible()

        editor.tool_bar.tool(tool="freehand_eraser").wait_for_and_click()
        self.mouse_drag(
            from_element=freehand_comment_b, from_offset=(100, -100), via_offsets=[(-100, 100)], to_offset=(-100, 100)
        )

        editor.comment.freehand_comment.wait_for_absent()

    @selenium_test
    def test_editor_snapping(self):
        editor = self.components.workflow_editor
        self.workflow_create_new(annotation="simple workflow")
        self.sleep_for(self.wait_types.UX_RENDER)

        self.tool_open("cat")
        self.sleep_for(self.wait_types.UX_RENDER)
        editor.label_input.wait_for_and_send_keys("tool_node")

        # activate snapping and set it to max (200)
        editor.tool_bar.tool(tool="toggle_snap").wait_for_and_click()
        editor.tool_bar.snapping_distance.wait_for_and_click()
        self.send_keys_to_page(Keys.RIGHT * 10)

        # move the node a bit
        tool_node = editor.node._(label="tool_node").wait_for_present()
        self.mouse_drag(from_element=tool_node, to_offset=(12, 3))

        # check if editor position is snapped
        top, left = self.get_node_position("tool_node")

        assert top % 200 == 0
        assert left % 200 == 0

        # move the node a bit more
        tool_node = editor.node._(label="tool_node").wait_for_present()
        self.mouse_drag(from_element=tool_node, to_offset=(207, -181))

        # check if editor position is snapped
        top, left = self.get_node_position("tool_node")

        assert top % 200 == 0
        assert left % 200 == 0

    @selenium_test
    def test_editor_selection(self):
        editor = self.components.workflow_editor
        self.workflow_create_new(annotation="simple workflow")
        self.sleep_for(self.wait_types.UX_RENDER)

        canvas = editor.canvas_body.wait_for_visible()

        # place tool in center of canvas
        self.workflow_editor_add_tool_step("cat")
        self.sleep_for(self.wait_types.UX_RENDER)
        editor.label_input.wait_for_and_send_keys("tool_node")
        tool_node = editor.node._(label="tool_node").wait_for_present()
        self.mouse_drag(from_element=tool_node, to_element=canvas, to_offset=(0, -100))

        # select the node
        editor.node_inspector_close.wait_for_and_click()
        self.shift_click(tool_node)
        self.sleep_for(self.wait_types.UX_RENDER)

        assert editor.tool_bar.selection_count.wait_for_visible().text.find("1 step") != -1

        # duplicate it
        editor.tool_bar.duplicate_selection.wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)

        assert editor.tool_bar.selection_count.wait_for_visible().text.find("1 step") != -1

        # move the node
        tool_node_new = editor.node._(label="tool_node 2").wait_for_present()
        self.mouse_drag(from_element=tool_node_new, to_element=canvas, to_offset=(0, 100))

        # clear selection
        editor.tool_bar.clear_selection.wait_for_and_click()
        editor.tool_bar.selection_count.wait_for_absent_or_hidden()

        # select both using box
        tool_node_original = editor.node._(label="tool_node").wait_for_present()
        editor.tool_bar.tool(tool="box_select").wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)

        self.mouse_drag(
            from_element=tool_node_original, from_offset=(150, -100), to_element=tool_node_new, to_offset=(-150, 100)
        )
        self.sleep_for(self.wait_types.UX_RENDER)

        assert editor.tool_bar.selection_count.wait_for_visible().text.find("2 steps") != -1

        # delete steps
        editor.tool_bar.delete_selection.wait_for_and_click()
        editor.tool_bar.selection_count.wait_for_absent_or_hidden()
        editor.node._(label="tool_node").wait_for_absent()
        editor.node._(label="tool_node 2").wait_for_absent()

        # place comments
        editor.tool_bar.tool(tool="text_comment").wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)
        self.mouse_drag(from_element=canvas, from_offset=(-100, -200), to_element=canvas, to_offset=(100, 0))

        editor.tool_bar.tool(tool="markdown_comment").wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)
        self.mouse_drag(from_element=canvas, from_offset=(-100, 0), to_element=canvas, to_offset=(100, 200))

        # select both using box select
        editor.tool_bar.tool(tool="box_select").wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)

        self.mouse_drag(from_element=canvas, from_offset=(-110, -210), to_element=canvas, to_offset=(110, 210))

        assert editor.tool_bar.selection_count.wait_for_visible().text.find("2 comments") != -1

        # deselect one using box select
        editor.tool_bar.select_mode_remove.wait_for_and_click()
        self.sleep_for(self.wait_types.UX_RENDER)

        self.mouse_drag(from_element=canvas, from_offset=(-110, -210), to_element=canvas, to_offset=(110, 10))
        self.sleep_for(self.wait_types.UX_RENDER)

        assert editor.tool_bar.selection_count.wait_for_visible().text.find("1 comment") != -1

    def create_and_wait_for_new_workflow_in_editor(self, annotation: str | None = None) -> str:
        editor = self.components.workflow_editor
        name = self.workflow_create_new(annotation=annotation)
        editor.canvas_body.wait_for_visible()
        return name

    def save_after_node_form_changes(self):
        # onSetData does an extra POST to build_modules, so we need to wait for that ...
        self.sleep_for(self.wait_types.UX_RENDER)
        self.assert_workflow_has_changes_and_save()

    def get_node_position(self, label: str):
        node = self.components.workflow_editor.node._(label=label).wait_for_present()

        return self.get_element_position(node)

    def get_element_position(self, element: WebElementProtocol):
        left = element.value_of_css_property("left")
        top = element.value_of_css_property("top")

        left_stripped = "".join(char for char in left if char.isdigit())
        top_stripped = "".join(char for char in top if char.isdigit())

        return (int(left_stripped), int(top_stripped))

    def get_element_size(self, element: WebElementProtocol):
        width = element.value_of_css_property("width")
        height = element.value_of_css_property("height")

        width_stripped = "".join(char for char in width if char.isdigit())
        height_stripped = "".join(char for char in height if char.isdigit())

        return (int(width_stripped), int(height_stripped))

    @retry_assertion_during_transitions
    def assert_node_output_is(self, label: str, output_type: str, subcollection_type: str | None = None):
        editor = self.components.workflow_editor
        node_label, output_name = label.split("#")
        node = editor.node._(label=node_label)
        node.wait_for_present()
        # Dismiss any stale tooltip before hovering so retry gets fresh text
        self.click_center()
        output_element = node.output_terminal(name=output_name).wait_for_visible()
        self.hover_over(output_element)
        element = self.components._.tooltip_inner.wait_for_present()
        assert f"output is {output_type}" in element.text, element.text
        if subcollection_type is None:
            assert "mapped-over" not in element.text
        else:
            fragment = " and mapped-over to produce a "
            if subcollection_type == "list:paired":
                fragment += "list of pairs dataset collection"
            elif subcollection_type == "list:list":
                fragment += "list of lists dataset collection"
            elif subcollection_type.count(":") > 1:
                fragment += f"dataset collection with {subcollection_type.count(':') + 1} levels of nesting"
            else:
                fragment += f"{subcollection_type}"
            assert fragment in element.text
        self.click_center()

    def workflow_editor_maximize_center_pane(self, collapse_left=True, collapse_right=True):
        self.sleep_for(self.wait_types.UX_RENDER)
        editor = self.components.workflow_editor

        if collapse_right and not editor.node_inspector.is_absent:
            editor.node_inspector_close.wait_for_and_click()
        if collapse_left and not self.components._.active_nav_item.is_absent:
            self.components._.active_nav_item.wait_for_and_click()

        self.sleep_for(self.wait_types.UX_RENDER)

    def assert_connected(self, source, sink):
        source_id, sink_id = self.workflow_editor_source_sink_terminal_ids(source, sink)
        # SVG <g> elements are considered "hidden" by Playwright even when
        # rendered, so use wait_for_present instead of wait_for_visible
        # with the Playwright backend.
        connector = self.components.workflow_editor.connector_for(source_id=source_id, sink_id=sink_id)
        if self._driver_impl.backend_type == "playwright":
            connector.wait_for_present()
        else:
            connector.wait_for_visible()

    def assert_connection_invalid(self, source, sink):
        source_id, sink_id = self.workflow_editor_source_sink_terminal_ids(source, sink)
        self.components.workflow_editor.connector_invalid_for(source_id=source_id, sink_id=sink_id).wait_for_present()

    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, auto_layout=True):
        name = self.workflow_upload_yaml_with_random_name(yaml_content)
        self.workflow_index_open()
        self.workflow_index_open_with_name(name)
        self.workflow_editor_dismiss_state_upgrade_modal()
        if auto_layout:
            self.components.workflow_editor.tool_bar.auto_layout.wait_for_and_click()
            self.sleep_for(self.wait_types.UX_RENDER)
        return name

    def workflow_editor_dismiss_state_upgrade_modal(self):
        """Dismiss the StateUpgradeModal if it appears when opening a workflow.

        This Bootstrap Vue modal intercepts pointer events and blocks clicks
        on the workflow editor canvas. Wait briefly for it to potentially
        appear, then dismiss if present.
        """
        editor = self.components.workflow_editor
        self.sleep_for(self.wait_types.UX_RENDER)
        if editor.state_upgrade_modal.is_displayed:
            editor.state_upgrade_modal_close.wait_for_and_click()
            editor.state_upgrade_modal.wait_for_absent_or_hidden()

    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).wait_for_visible()
        self.hover_over(sink_input)
        sink_node.connector_destroy_callout(name=sink_input_name).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()

    @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}'"

    @retry_assertion_during_transitions
    def assert_modal_has_text(self, expected_text):
        modal_element = self.components.workflow_editor.state_upgrade_modal.wait_for_visible()
        text = modal_element.text
        assert expected_text in text, f"Failed to find expected text [{expected_text}] in modal text [{text}]"

    def move_center_of_canvas(self, xoffset=0, yoffset=0):
        _canvas = self.find_element_by_id("canvas-container")
        if self.backend_type == "playwright":
            pw_driver = cast("HasPlaywrightDriver", self._driver_impl)
            page = pw_driver.page
            handle = pw_driver._unwrap_element(_canvas)
            box = handle.bounding_box()
            assert box is not None
            cx = box["x"] + box["width"] / 2
            cy = box["y"] + box["height"] / 2
            page.mouse.move(cx, cy)
            page.mouse.down()
            page.mouse.move(cx + xoffset, cy + yoffset)
            page.mouse.up()
        else:
            canvas = cast(WebElement, _canvas)
            chains = ActionChains(self.driver)
            chains.click_and_hold(canvas).move_by_offset(xoffset=xoffset, yoffset=yoffset).release().perform()