diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ccf1a1..4cc7c8d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,5 +8,3 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added - -- `graph_to_svg`: Generates SVG mock-ups from Ewoks workflows. diff --git a/examples/convert_ewoks_to_elk_graph.py b/examples/convert_ewoks_to_elk_graph.py index 32d30b4..c0ec650 100644 --- a/examples/convert_ewoks_to_elk_graph.py +++ b/examples/convert_ewoks_to_elk_graph.py @@ -7,6 +7,8 @@ from ewoksdraw.layout.elk_converter import ElkGraph from ewoksdraw.layout.elk_converter import convert_ewoks_to_elk_graph from ewoksdraw.svg.svg_task_group import SvgTaskGroup +from ewoksdraw.svg.svg_task_group import TaskInputPositions +from ewoksdraw.svg.svg_task_group import TaskOutputPositions from ewoksdraw.svg.svg_task_group import TaskSizes graph_description, _ = get_graph("acyclic1") @@ -14,9 +16,15 @@ svg_task_group: SvgTaskGroup = build_svg_task_group(ewoks_graph) task_sizes: TaskSizes = svg_task_group.extract_task_sizes() -elk_graph: ElkGraph = convert_ewoks_to_elk_graph(ewoks_graph, task_sizes) +task_input_positions: TaskInputPositions = svg_task_group.extract_input_positions() +task_output_positions: TaskOutputPositions = svg_task_group.extract_output_positions() +elk_graph: ElkGraph = convert_ewoks_to_elk_graph( + ewoks_graph, task_sizes, task_input_positions, task_output_positions +) pprint(dict(ewoks_graph.graph.nodes(data=True))) pprint(list(ewoks_graph.graph.edges(data=True))) pprint(task_sizes) +pprint(task_input_positions) +pprint(task_output_positions) pprint(elk_graph) diff --git a/src/ewoksdraw/layout/elk_converter.py b/src/ewoksdraw/layout/elk_converter.py index 80b4e05..20328e4 100644 --- a/src/ewoksdraw/layout/elk_converter.py +++ b/src/ewoksdraw/layout/elk_converter.py @@ -1,3 +1,4 @@ +import warnings from typing import Any from typing import TypedDict @@ -5,13 +6,27 @@ from ewoksdraw.config.constants import ELK_LAYOUT_OPTIONS +from ..svg.svg_task import TaskIOPosition +from ..svg.svg_task_group import TaskInputPositions +from ..svg.svg_task_group import TaskOutputPositions from ..svg.svg_task_group import TaskSizes +class ElkPort(TypedDict): + id: str + x: float + y: float + width: float + height: float + layoutOptions: dict[str, Any] + + class ElkChild(TypedDict): id: str width: float height: float + layoutOptions: dict[str, Any] + ports: list[ElkPort] class ElkEdge(TypedDict): @@ -28,13 +43,17 @@ class ElkGraph(TypedDict): def convert_ewoks_to_elk_graph( - ewoks_graph: TaskGraph, task_sizes: TaskSizes + ewoks_graph: TaskGraph, + task_sizes: TaskSizes, + task_input_positions: TaskInputPositions, + task_output_positions: TaskOutputPositions, ) -> ElkGraph: """Convert an Ewoks task graph into an ELK layout graph. :param ewoks_graph: the task graph to convert, e.g. from ``ewokscore.load_graph``. - :param task_sizes: ``(width, height)`` per task in ``ewoks_graph``, and no - other task id, e.g. ``{"task1": (39.56, 55.0), ...}``. + :param task_sizes: width and height of each task. + :param task_input_positions: input positions of each task. + :param task_output_positions: output positions of each task. """ node_ids = set(ewoks_graph.graph.nodes) if node_ids != task_sizes.keys(): @@ -43,30 +62,135 @@ def convert_ewoks_to_elk_graph( f"{sorted(node_ids)}" ) - children: list[ElkChild] = [ - { - "id": task_id, - "width": task_sizes[task_id].width, - "height": task_sizes[task_id].height, - } - for task_id in ewoks_graph.graph.nodes - ] + if node_ids != task_input_positions.keys(): + raise ValueError( + "task_input_positions " + f"{sorted(task_input_positions)} do not match ewoks_graph task ids " + f"{sorted(node_ids)}" + ) + + if node_ids != task_output_positions.keys(): + raise ValueError( + "task_output_positions " + f"{sorted(task_output_positions)} do not match ewoks_graph task ids " + f"{sorted(node_ids)}" + ) + + children: list[ElkChild] = [] + used_ids: set[str] = set() + for task_id in ewoks_graph.graph.nodes: + ports = _convert_io_positions_to_elk_ports( + task_id, + task_input_positions[task_id], + task_output_positions[task_id], + ) + children.append( + { + "id": task_id, + "width": task_sizes[task_id].width, + "height": task_sizes[task_id].height, + "layoutOptions": { + "org.eclipse.elk.portConstraints": "FIXED_POS", + }, + "ports": ports, + } + ) + used_ids.add(task_id) + used_ids.update(port["id"] for port in ports) + + root_id = _available_elk_id("__ewoksdraw_root__", used_ids) + used_ids.add(root_id) edges: list[ElkEdge] = [] for source, target, link_attrs in ewoks_graph.graph.edges(data=True): - n_mappings = len(link_attrs.get("data_mapping", [])) or 1 - for index in range(n_mappings): + if link_attrs.get("map_all_data", False): + warnings.warn( + f"Ewoks link {source!r} -> {target!r} uses 'map_all_data', which " + "is not yet supported.", + UserWarning, + stacklevel=2, + ) + + for mapping in link_attrs.get("data_mapping", []): + source_output = mapping.get("source_output") + if source_output is None: + warnings.warn( + f"Data mapping on Ewoks link {source!r} -> {target!r} has no " + "'source_output', which is not yet supported.", + UserWarning, + stacklevel=2, + ) + continue + + edge_id = _available_elk_id( + f"edge_{len(edges)}_{source}_{target}", used_ids + ) + edges.append( { - "id": f"edge_{source}_{target}_{index}", - "sources": [source], - "targets": [target], + "id": edge_id, + "sources": [f"{source}.output.{source_output}"], + "targets": [f"{target}.input.{mapping['target_input']}"], } ) + used_ids.add(edge_id) return { - "id": "root", + "id": root_id, "layoutOptions": ELK_LAYOUT_OPTIONS, "children": children, "edges": edges, } + + +def _convert_io_positions_to_elk_ports( + task_id: str, + input_positions: list[TaskIOPosition], + output_positions: list[TaskIOPosition], +) -> list[ElkPort]: + ports = _convert_positions_to_elk_ports( + input_positions, + id_prefix=f"{task_id}.input", + elk_port_side="WEST", + ) + ports.extend( + _convert_positions_to_elk_ports( + output_positions, + id_prefix=f"{task_id}.output", + elk_port_side="EAST", + ) + ) + return ports + + +def _convert_positions_to_elk_ports( + positions: list[TaskIOPosition], id_prefix: str, elk_port_side: str +) -> list[ElkPort]: + ports: list[ElkPort] = [] + + for index, position in enumerate(positions): + ports.append( + { + "id": f"{id_prefix}.{position.name}", + "x": position.x, + "y": position.y, + "width": 0, + "height": 0, + "layoutOptions": { + "org.eclipse.elk.port.side": elk_port_side, + "org.eclipse.elk.port.index": index, + "org.eclipse.elk.port.borderOffset": 0, + }, + } + ) + + return ports + + +def _available_elk_id(preferred_id: str, used_ids: set[str]) -> str: + element_id = preferred_id + index = 1 + while element_id in used_ids: + element_id = f"{preferred_id}_{index}" + index += 1 + return element_id diff --git a/src/ewoksdraw/svg/svg_group.py b/src/ewoksdraw/svg/svg_group.py index 6489067..feda6f3 100644 --- a/src/ewoksdraw/svg/svg_group.py +++ b/src/ewoksdraw/svg/svg_group.py @@ -1,6 +1,7 @@ import re from typing import Generic from typing import Iterable +from typing import NamedTuple from typing import Protocol from typing import TypeVar from xml.etree.ElementTree import Element @@ -14,6 +15,11 @@ def xml_element(self) -> Element: ... SvgElementType = TypeVar("SvgElementType", bound=SvgElementLike) +class Translation(NamedTuple): + x: float + y: float + + class SvgGroup(Generic[SvgElementType]): """ Represents a group of SVG elements. @@ -27,6 +33,12 @@ def __init__(self, group_id: str | None = None): self.elements: list[SvgElementType] = [] self._group_id = group_id self._transform = "" + self._translation = Translation(x=0.0, y=0.0) + + @property + def translation(self) -> Translation: + """Returns the current translation as an ``(x, y)`` tuple.""" + return self._translation def add_elements(self, elements: Iterable[SvgElementType]) -> None: """ @@ -49,6 +61,8 @@ def translate(self, x: float = 0, y: float = 0) -> None: else: self._transform = new_transform self._transform = self._transform.strip() + current_x, current_y = self._translation + self._translation = Translation(x=current_x + x, y=current_y + y) def set_translation(self, x: float = 0, y: float = 0) -> None: """ @@ -66,6 +80,7 @@ def set_translation(self, x: float = 0, y: float = 0) -> None: self._transform = f"{cleaned_transform} {new_translate}".strip() else: self._transform = new_translate + self._translation = Translation(x=x, y=y) @property def xml_element(self) -> Element: diff --git a/src/ewoksdraw/svg/svg_task.py b/src/ewoksdraw/svg/svg_task.py index 8d5c3e1..421187e 100644 --- a/src/ewoksdraw/svg/svg_task.py +++ b/src/ewoksdraw/svg/svg_task.py @@ -14,6 +14,12 @@ class TaskSize(NamedTuple): height: float +class TaskIOPosition(NamedTuple): + name: str + x: float + y: float + + class SvgTask(SvgGroup): """ Represents a task as an SVG group containing title, input/output groups, box, and @@ -35,6 +41,7 @@ def __init__( ): super().__init__(group_id=task_name) + self._task_name = task_name self._interspace_title_input = IO_TOP_MARGIN self._interspace_input_output = IO_INTER_IO_MARGIN self._title = SvgTaskTitle(text=task_name, x=0, y=0) @@ -131,3 +138,20 @@ def height(self) -> float: + self._interspace_input_output + self._interspace_title_input ) + + def get_input_positions(self) -> list[TaskIOPosition]: + """Return the task-relative positions of the task inputs.""" + return self._get_group_io_positions(self._inputs) + + def get_output_positions(self) -> list[TaskIOPosition]: + """Return the task-relative positions of the task outputs.""" + return self._get_group_io_positions(self._outputs) + + @staticmethod + def _get_group_io_positions(group: SvgTaskIOGroup) -> list[TaskIOPosition]: + positions: list[TaskIOPosition] = [] + for io in group.elements: + x = group.translation.x + io.translation.x + y = group.translation.y + io.translation.y + positions.append(TaskIOPosition(name=io.name, x=x, y=y)) + return positions diff --git a/src/ewoksdraw/svg/svg_task_group.py b/src/ewoksdraw/svg/svg_task_group.py index ecd1e24..5a011c0 100644 --- a/src/ewoksdraw/svg/svg_task_group.py +++ b/src/ewoksdraw/svg/svg_task_group.py @@ -1,8 +1,11 @@ from .svg_group import SvgGroup from .svg_task import SvgTask +from .svg_task import TaskIOPosition from .svg_task import TaskSize TaskSizes = dict[str, TaskSize] +TaskInputPositions = dict[str, list[TaskIOPosition]] +TaskOutputPositions = dict[str, list[TaskIOPosition]] class SvgTaskGroup(SvgGroup[SvgTask]): @@ -53,3 +56,15 @@ def extract_task_sizes(self) -> TaskSizes: task_id: TaskSize(width=svg_task.width, height=svg_task.height) for task_id, svg_task in self._svg_tasks.items() } + + def extract_input_positions(self) -> TaskInputPositions: + return { + task_id: svg_task.get_input_positions() + for task_id, svg_task in self._svg_tasks.items() + } + + def extract_output_positions(self) -> TaskOutputPositions: + return { + task_id: svg_task.get_output_positions() + for task_id, svg_task in self._svg_tasks.items() + } diff --git a/src/ewoksdraw/svg/svg_task_io.py b/src/ewoksdraw/svg/svg_task_io.py index b8d862c..79eca92 100644 --- a/src/ewoksdraw/svg/svg_task_io.py +++ b/src/ewoksdraw/svg/svg_task_io.py @@ -36,6 +36,15 @@ def set_font_size(self, font_size: float) -> None: """ self.txt.set_font_size(font_size) + @property + def name(self) -> str: + """Return the original, untruncated task input or output name.""" + return self._io_txt + + @property + def io_type(self) -> Literal["input", "output"]: + return self._io_type + @property def font_size(self) -> float: """ diff --git a/src/ewoksdraw/tests/test_elk_converter.py b/src/ewoksdraw/tests/test_elk_converter.py index 40565ac..23bd9fd 100644 --- a/src/ewoksdraw/tests/test_elk_converter.py +++ b/src/ewoksdraw/tests/test_elk_converter.py @@ -5,8 +5,12 @@ from ewokscore.tests.examples.graphs import graph_names from pyelk.graph import validate_graph +from ewoksdraw import build_svg_task_group from ewoksdraw.config.constants import ELK_LAYOUT_OPTIONS from ewoksdraw.layout.elk_converter import convert_ewoks_to_elk_graph +from ewoksdraw.svg.svg_task import TaskIOPosition +from ewoksdraw.svg.svg_task_group import TaskInputPositions +from ewoksdraw.svg.svg_task_group import TaskOutputPositions from ewoksdraw.svg.svg_task_group import TaskSize from ewoksdraw.svg.svg_task_group import TaskSizes @@ -24,24 +28,62 @@ def _task_sizes(graph: TaskGraph) -> TaskSizes: } +def _task_input_positions(graph: TaskGraph) -> TaskInputPositions: + return {node_id: [] for node_id in graph.graph.nodes} + + +def _task_output_positions(graph: TaskGraph) -> TaskOutputPositions: + return {node_id: [] for node_id in graph.graph.nodes} + + def test_top_level_structure() -> None: graph_description, _ = get_graph("acyclic1") graph = load_graph(graph_description) - elk_graph = convert_ewoks_to_elk_graph(graph, _task_sizes(graph)) + elk_graph = convert_ewoks_to_elk_graph( + graph, + _task_sizes(graph), + _task_input_positions(graph), + _task_output_positions(graph), + ) - assert elk_graph["id"] == "root" + assert elk_graph["id"] == "__ewoksdraw_root__" assert elk_graph["layoutOptions"] == ELK_LAYOUT_OPTIONS assert "children" in elk_graph assert "edges" in elk_graph +def test_root_id_does_not_collide_with_task_ids() -> None: + task_id = "__ewoksdraw_root__" + graph_description = { + "graph": {"id": "g", "label": "g", "schema_version": "1.1"}, + "nodes": [_node(task_id)], + "links": [], + } + graph = load_graph(graph_description) + + elk_graph = convert_ewoks_to_elk_graph( + graph, + {task_id: TaskSize(width=1.0, height=1.0)}, + {task_id: []}, + {task_id: []}, + ) + + assert elk_graph["id"] == "__ewoksdraw_root___1" + assert elk_graph["children"][0]["id"] == task_id + + def test_children_match_task_sizes() -> None: graph_description, _ = get_graph("acyclic1") graph = load_graph(graph_description) task_sizes = _task_sizes(graph) - elk_graph = convert_ewoks_to_elk_graph(graph, task_sizes) + elk_graph = convert_ewoks_to_elk_graph( + graph, + task_sizes, + _task_input_positions(graph), + _task_output_positions(graph), + ) assert len(elk_graph["children"]) == len(task_sizes) for child in elk_graph["children"]: @@ -50,7 +92,52 @@ def test_children_match_task_sizes() -> None: assert child["height"] == height -def test_link_without_data_mapping_produces_one_elk_edge() -> None: +def test_children_include_io_positions_as_elk_ports() -> None: + graph_description = { + "graph": {"id": "g", "label": "g", "schema_version": "1.1"}, + "nodes": [_node("task")], + "links": [], + } + graph = load_graph(graph_description) + input_positions = {"task": [TaskIOPosition(name="value", x=0.0, y=10.0)]} + output_positions = {"task": [TaskIOPosition(name="result", x=20.0, y=15.0)]} + + elk_graph = convert_ewoks_to_elk_graph( + graph, + {"task": TaskSize(width=20.0, height=30.0)}, + input_positions, + output_positions, + ) + + assert elk_graph["children"][0]["ports"] == [ + { + "id": "task.input.value", + "x": 0.0, + "y": 10.0, + "width": 0, + "height": 0, + "layoutOptions": { + "org.eclipse.elk.port.side": "WEST", + "org.eclipse.elk.port.index": 0, + "org.eclipse.elk.port.borderOffset": 0, + }, + }, + { + "id": "task.output.result", + "x": 20.0, + "y": 15.0, + "width": 0, + "height": 0, + "layoutOptions": { + "org.eclipse.elk.port.side": "EAST", + "org.eclipse.elk.port.index": 0, + "org.eclipse.elk.port.borderOffset": 0, + }, + }, + ] + + +def test_link_without_data_mapping_produces_no_elk_edge() -> None: graph_description = { "graph": {"id": "g", "label": "g", "schema_version": "1.1"}, "nodes": [_node("a"), _node("b")], @@ -59,10 +146,90 @@ def test_link_without_data_mapping_produces_one_elk_edge() -> None: graph = load_graph(graph_description) size = TaskSize(width=1.0, height=1.0) - elk_graph = convert_ewoks_to_elk_graph(graph, {"a": size, "b": size}) + elk_graph = convert_ewoks_to_elk_graph( + graph, + {"a": size, "b": size}, + {"a": [], "b": []}, + {"a": [], "b": []}, + ) + + assert elk_graph["edges"] == [] + + +def test_link_with_map_all_data_produces_no_elk_edge() -> None: + graph_description = { + "graph": {"id": "g", "label": "g", "schema_version": "1.1"}, + "nodes": [_node("a"), _node("b")], + "links": [{"source": "a", "target": "b", "map_all_data": True}], + } + graph = load_graph(graph_description) + + size = TaskSize(width=1.0, height=1.0) + with pytest.warns( + UserWarning, + match=( + r"Ewoks link 'a' -> 'b' uses 'map_all_data', which is not yet supported\." + ), + ): + elk_graph = convert_ewoks_to_elk_graph( + graph, + {"a": size, "b": size}, + {"a": [], "b": []}, + {"a": [], "b": []}, + ) + + assert elk_graph["edges"] == [] + + +def test_data_mapping_without_source_output_is_not_drawn() -> None: + graph_description = { + "graph": {"id": "g", "label": "g", "schema_version": "1.1"}, + "nodes": [_node("a"), _node("b")], + "links": [ + { + "source": "a", + "target": "b", + "data_mapping": [ + {"target_input": "all_results"}, + {"source_output": "result", "target_input": "value"}, + ], + } + ], + } + graph = load_graph(graph_description) + size = TaskSize(width=1.0, height=1.0) + input_positions = { + "a": [], + "b": [ + TaskIOPosition(name="all_results", x=0.0, y=0.25), + TaskIOPosition(name="value", x=0.0, y=0.75), + ], + } + output_positions = { + "a": [TaskIOPosition(name="result", x=1.0, y=0.5)], + "b": [], + } + + with pytest.warns( + UserWarning, + match=( + r"Data mapping on Ewoks link 'a' -> 'b' has no 'source_output', " + r"which is not yet supported\." + ), + ): + elk_graph = convert_ewoks_to_elk_graph( + graph, + {"a": size, "b": size}, + input_positions, + output_positions, + ) assert elk_graph["edges"] == [ - {"id": "edge_a_b_0", "sources": ["a"], "targets": ["b"]} + { + "id": "edge_0_a_b", + "sources": ["a.output.result"], + "targets": ["b.input.value"], + } ] @@ -84,36 +251,102 @@ def test_link_with_multiple_data_mappings_produces_one_elk_edge_per_mapping() -> graph = load_graph(graph_description) size = TaskSize(width=1.0, height=1.0) - elk_graph = convert_ewoks_to_elk_graph(graph, {"a": size, "b": size}) + input_positions = { + "a": [], + "b": [ + TaskIOPosition(name="a", x=0.0, y=0.25), + TaskIOPosition(name="b", x=0.0, y=0.75), + ], + } + output_positions = { + "a": [TaskIOPosition(name="result", x=1.0, y=0.5)], + "b": [], + } + elk_graph = convert_ewoks_to_elk_graph( + graph, + {"a": size, "b": size}, + input_positions, + output_positions, + ) assert elk_graph["edges"] == [ - {"id": "edge_a_b_0", "sources": ["a"], "targets": ["b"]}, - {"id": "edge_a_b_1", "sources": ["a"], "targets": ["b"]}, + { + "id": "edge_0_a_b", + "sources": ["a.output.result"], + "targets": ["b.input.a"], + }, + { + "id": "edge_1_a_b", + "sources": ["a.output.result"], + "targets": ["b.input.b"], + }, ] +def test_edge_id_does_not_collide_with_task_ids() -> None: + colliding_task_id = "edge_0_a_b" + graph_description = { + "graph": {"id": "g", "label": "g", "schema_version": "1.1"}, + "nodes": [_node("a"), _node("b"), _node(colliding_task_id)], + "links": [ + { + "source": "a", + "target": "b", + "data_mapping": [{"source_output": "result", "target_input": "value"}], + } + ], + } + graph = load_graph(graph_description) + size = TaskSize(width=1.0, height=1.0) + input_positions = { + "a": [], + "b": [TaskIOPosition(name="value", x=0.0, y=0.5)], + colliding_task_id: [], + } + output_positions = { + "a": [TaskIOPosition(name="result", x=1.0, y=0.5)], + "b": [], + colliding_task_id: [], + } + + elk_graph = convert_ewoks_to_elk_graph( + graph, + {"a": size, "b": size, colliding_task_id: size}, + input_positions, + output_positions, + ) + + assert elk_graph["edges"][0]["id"] == "edge_0_a_b_1" + + def test_graph_without_link() -> None: graph_description, _ = get_graph("empty") graph = load_graph(graph_description) - elk_graph = convert_ewoks_to_elk_graph(graph, {}) + elk_graph = convert_ewoks_to_elk_graph(graph, {}, {}, {}) assert elk_graph["children"] == [] assert elk_graph["edges"] == [] @pytest.mark.parametrize("graph_name", graph_names()) +@pytest.mark.filterwarnings("ignore:.*uses 'map_all_data'.*:UserWarning") def test_children_and_edges_count_across_example_graphs(graph_name: str) -> None: graph_description, _ = get_graph(graph_name) graph = load_graph(graph_description) task_sizes = _task_sizes(graph) - elk_graph = convert_ewoks_to_elk_graph(graph, task_sizes) + elk_graph = convert_ewoks_to_elk_graph( + graph, + task_sizes, + _task_input_positions(graph), + _task_output_positions(graph), + ) assert len(elk_graph["children"]) == graph.graph.number_of_nodes() expected_edge_count = sum( - len(link_attrs.get("data_mapping") or []) or 1 + len(link_attrs.get("data_mapping") or []) for _, _, link_attrs in graph.graph.edges(data=True) ) assert len(elk_graph["edges"]) == expected_edge_count @@ -124,7 +357,12 @@ def test_task_sizes_missing_a_node_raises() -> None: graph = load_graph(graph_description) with pytest.raises(ValueError): - convert_ewoks_to_elk_graph(graph, {"task1": TaskSize(width=10.0, height=20.0)}) + convert_ewoks_to_elk_graph( + graph, + {"task1": TaskSize(width=10.0, height=20.0)}, + _task_input_positions(graph), + _task_output_positions(graph), + ) def test_task_sizes_with_extra_task_id_raises() -> None: @@ -134,17 +372,54 @@ def test_task_sizes_with_extra_task_id_raises() -> None: task_sizes["not_a_node"] = TaskSize(width=1.0, height=1.0) with pytest.raises(ValueError): - convert_ewoks_to_elk_graph(graph, task_sizes) + convert_ewoks_to_elk_graph( + graph, + task_sizes, + _task_input_positions(graph), + _task_output_positions(graph), + ) + + +def test_task_input_positions_missing_a_node_raises() -> None: + graph_description, _ = get_graph("acyclic1") + graph = load_graph(graph_description) + + with pytest.raises(ValueError, match="task_input_positions"): + convert_ewoks_to_elk_graph( + graph, + _task_sizes(graph), + {"task1": []}, + _task_output_positions(graph), + ) + + +def test_task_output_positions_missing_a_node_raises() -> None: + graph_description, _ = get_graph("acyclic1") + graph = load_graph(graph_description) + + with pytest.raises(ValueError, match="task_output_positions"): + convert_ewoks_to_elk_graph( + graph, + _task_sizes(graph), + _task_input_positions(graph), + {"task1": []}, + ) @pytest.mark.parametrize("graph_name", graph_names()) +@pytest.mark.filterwarnings("ignore:.*uses 'map_all_data'.*:UserWarning") def test_output_is_a_valid_pyelk_graph(graph_name: str) -> None: """Check graph validation from pyelk""" graph_description, _ = get_graph(graph_name) graph = load_graph(graph_description) - task_sizes = _task_sizes(graph) + task_group = build_svg_task_group(graph) - elk_graph = convert_ewoks_to_elk_graph(graph, task_sizes) + elk_graph = convert_ewoks_to_elk_graph( + graph, + task_group.extract_task_sizes(), + task_group.extract_input_positions(), + task_group.extract_output_positions(), + ) validate_graph(elk_graph) diff --git a/src/ewoksdraw/tests/test_svg_task.py b/src/ewoksdraw/tests/test_svg_task.py new file mode 100644 index 0000000..b3255c5 --- /dev/null +++ b/src/ewoksdraw/tests/test_svg_task.py @@ -0,0 +1,16 @@ +from ewoksdraw.svg.svg_task import SvgTask + + +def test_io_positions_keep_original_names() -> None: + long_input_name = "input_name_" * 30 + task = SvgTask( + task_name="task", + input_names=[long_input_name], + output_names=["result"], + ) + + input_positions = task.get_input_positions() + output_positions = task.get_output_positions() + + assert [position.name for position in input_positions] == [long_input_name] + assert [position.name for position in output_positions] == ["result"]