`` label for a leaf node box."""
+ fill, line = style.NODE_PALETTE[node.kind]
+ rows = [f'| {_title_cell(node)} |
']
+ row_count = max(len(node.inputs), len(node.outputs))
+ for i in range(row_count):
+ left = _port_cell(node.inputs[i], "i") if i < len(node.inputs) else " | "
+ right = (
+ _port_cell(node.outputs[i], "o") if i < len(node.outputs) else " | "
+ )
+ rows.append(f'{left} | {right}
')
+ table = (
+ f''
+ )
+ return f"<{table}>"
+
+
+def _title_cell(node: model.DrawNode) -> str:
+ """The bold, wrapped label plus the optional monospace subtitle."""
+ lines = style.wrap_label(node.label)
+ html = f"{'
'.join(_escape(line) for line in lines)}"
+ if node.subtitle is not None:
+ html += (
+ f'
'
+ f"{_escape(node.subtitle)}"
+ )
+ return html
+
+
+def _port_content(port: model.DrawPort) -> str:
+ """The (possibly italicised, possibly badged) label content of a port cell."""
+ label = _escape(style.truncate_right(port.label, style.PORT_LABEL_MAX))
+ if port.has_default:
+ label = f"{label}"
+ extra = port.badge if port.badge is not None else port.hint
+ if extra is not None:
+ label += (
+ f'
{_escape(extra)}'
+ )
+ return label
+
+
+def _port_cell(port: model.DrawPort, prefix: str) -> str:
+ """A single rounded, grey-filled port cell, italicised and badged as needed."""
+ return (
+ f''
+ f"{_port_content(port)} | "
+ )
+
+
+def _emit_composite(context: graphviz.Digraph, node: model.DrawNode) -> None:
+ """Draw *node* as a filled cluster: its own IO boxes, its children, its edges."""
+ with context.subgraph(name=f"cluster_{node.path}") as sub:
+ fill, line = style.NODE_PALETTE[node.kind]
+ sub.attr(
+ label=_cluster_label(node),
+ style="rounded,filled",
+ fillcolor=fill,
+ color=line,
+ penwidth="2",
+ margin="12",
+ )
+ _emit_io_boxes(sub, node, node.inputs, base_models.IOTypes.INPUTS)
+ _emit_io_boxes(sub, node, node.outputs, base_models.IOTypes.OUTPUTS)
+ _emit_children(sub, node)
+ _emit_edges(sub, node)
+
+
+def _cluster_label(node: model.DrawNode) -> str:
+ """The cluster's own title, mirroring a leaf's title cell, plus an optional note."""
+ html = _title_cell(node)
+ if node.note is not None:
+ html += f"
{_escape(node.note)}"
+ return f"<{html}>"
+
+
+def _emit_io_boxes(
+ context: graphviz.Digraph,
+ node: model.DrawNode,
+ ports: tuple[model.DrawPort, ...],
+ io_type: base_models.IOTypes,
+) -> None:
+ """One small box per port in *ports*, rank-aligned together when there is more than one."""
+ if not ports:
+ return
+ if len(ports) > 1:
+ with context.subgraph() as rank_group:
+ rank_group.attr(rank="same")
+ for port in ports:
+ _emit_io_box(rank_group, node, port, io_type)
+ else:
+ _emit_io_box(context, node, ports[0], io_type)
+
+
+def _emit_io_box(
+ context: graphviz.Digraph,
+ node: model.DrawNode,
+ port: model.DrawPort,
+ io_type: base_models.IOTypes,
+) -> None:
+ """A composite's own IO port, drawn as a single-cell grey box just inside the wall."""
+ box_id = model.PortRef(node.path, io_type, port.label).lexical_path
+ table = (
+ f''
+ f'| {_port_content(port)} |
'
+ )
+ context.node(box_id, label=f"<{table}>")
+
+
+def _emit_children(context: graphviz.Digraph, node: model.DrawNode) -> None:
+ """Draw every child, nesting grouped ones inside a dashed group cluster.
+
+ Groups are emitted in reverse declaration order: under ``rankdir=LR``,
+ Graphviz stacks same-rank clusters bottom-up in declaration order, so
+ emitting the first-declared group last places it at the top, matching
+ evaluation order.
+ """
+ by_label = {lexical.split(child.path)[-1]: child for child in node.children}
+ grouped_paths: set[str] = set()
+ for index, group in reversed(list(enumerate(node.groups))):
+ group_name = lexical.join(node.path, f"group_{index}")
+ with context.subgraph(name=f"cluster_{group_name}") as group_sub:
+ group_sub.attr(
+ label=f"<{_escape(group.label)}>",
+ fontsize=_GROUP_FONT_SIZE,
+ style="rounded,dashed",
+ color=style.GROUP_LINE,
+ )
+ for member in group.members:
+ child = by_label[member]
+ _emit_node(group_sub, child)
+ grouped_paths.add(child.path)
+ for child in node.children:
+ if child.path not in grouped_paths:
+ _emit_node(context, child)
+
+
+def _emit_edges(context: graphviz.Digraph, node: model.DrawNode) -> None:
+ """Draw every edge of *node*, cutting the ones that would close a cycle."""
+ children_by_path = {child.path: child for child in node.children}
+ back_edges = _back_edges(node.edges)
+ for edge in node.edges:
+ tail = _endpoint(node, children_by_path, edge.source, compass="e")
+ head = _endpoint(node, children_by_path, edge.target, compass="w")
+ attrs: dict[str, str] = {}
+ if edge.conditional:
+ attrs["style"] = "dashed"
+ attrs["color"] = style.CONDITIONAL_LINE
+ if edge in back_edges:
+ attrs["constraint"] = "false"
+ context.edge(tail, head, **attrs)
+
+
+def _endpoint(
+ composite: model.DrawNode,
+ children_by_path: dict[str, model.DrawNode],
+ ref: model.PortRef,
+ *,
+ compass: str,
+) -> str:
+ """The ``node:port:compass`` string addressing *ref* from inside *composite*.
+
+ An empty ``node_path`` addresses *composite*'s own IO box. Otherwise it
+ addresses a real child: a leaf child's own port cell, or -- when the
+ child is itself an expanded composite -- that child's own IO box, which
+ is as close as Graphviz gets to piercing a cluster wall.
+ """
+ if not ref.node_path:
+ box_id = model.PortRef(composite.path, ref.io_type, ref.port).lexical_path
+ return f"{box_id}:p:{compass}"
+ child_path = lexical.join(composite.path, ref.node_path)
+ child = children_by_path[child_path]
+ if child.is_leaf:
+ prefix = "i" if ref.io_type is base_models.IOTypes.INPUTS else "o"
+ return f"{child_path}:{prefix}_{ref.port}:{compass}"
+ box_id = model.PortRef(child_path, ref.io_type, ref.port).lexical_path
+ return f"{box_id}:p:{compass}"
+
+
+def _back_edges(edges: tuple[model.DrawEdge, ...]) -> set[model.DrawEdge]:
+ """Depth-first cycle detection over *edges*, keyed by sibling node path.
+
+ ``while`` back-edges make a cluster's own edge set cyclic, which would
+ otherwise destroy ``rankdir=LR`` layout. This is a rendering-local layout
+ concern, not represented in the IR.
+
+ A real child collapses its input and output pins into one node identity,
+ since the dependency that matters here is "does this child's output feed
+ back into one of its own ancestors". A composite's own IO does not
+ collapse that way: its inputs and outputs are the source and sink of the
+ whole cluster, and treating them as one node would flag every ordinary
+ input-to-output path as a cycle.
+ """
+ adjacency: dict[str, list[tuple[str, model.DrawEdge]]] = {}
+ for edge in edges:
+ adjacency.setdefault(_node_key(edge.source), []).append(
+ (_node_key(edge.target), edge)
+ )
+ all_nodes = {_node_key(edge.source) for edge in edges} | {
+ _node_key(edge.target) for edge in edges
+ }
+ visited: set[str] = set()
+ in_progress: set[str] = set()
+ back: set[model.DrawEdge] = set()
+
+ def visit(current: str) -> None:
+ visited.add(current)
+ in_progress.add(current)
+ for target, edge in adjacency.get(current, []):
+ if target in in_progress:
+ back.add(edge)
+ elif target not in visited:
+ visit(target)
+ in_progress.discard(current)
+
+ for start in all_nodes:
+ if start not in visited:
+ visit(start)
+ return back
+
+
+def _node_key(ref: model.PortRef) -> str:
+ """The sibling-level node identity of *ref*, for cycle detection.
+
+ A real child's own label, ignoring which of its ports is referenced; a
+ composite's own input and its own output are distinct pseudo-nodes.
+ """
+ if ref.node_path:
+ return ref.node_path
+ return f"<{ref.io_type}>"
diff --git a/src/flowrep/drawing/retrospective.py b/src/flowrep/drawing/retrospective.py
new file mode 100644
index 00000000..fc38387a
--- /dev/null
+++ b/src/flowrep/drawing/retrospective.py
@@ -0,0 +1,161 @@
+"""
+Build a graphviz-free :class:`~flowrep.drawing.model.DrawGraph` from a
+retrospective :class:`~flowrep.retrospective.datastructures.NodeData`.
+
+Free of any graphviz import, so building a drawing works in a bare install.
+Every :class:`~flowrep.retrospective.datastructures.CompositeData` -- a run
+``DagData`` as much as a run ``ForEachData``, ``IfData``, ``TryData`` or
+``WhileData`` -- reads its ``.nodes``, ``.input_edges``, ``.edges`` and
+``.output_edges`` identically. There is no per-flow-control-type branching;
+the only dispatch is leaf versus composite.
+
+This module shares nothing with :mod:`flowrep.drawing.prospective` but the IR
+(:mod:`flowrep.drawing.model`) and the shared label formatting in
+:mod:`flowrep.drawing.style`.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+
+from flowrep import base_models, edge_models, lexical
+from flowrep.drawing import model, style
+from flowrep.retrospective import datastructures
+
+
+def build(data: datastructures.NodeData, depth: int = 0) -> model.DrawGraph:
+ """Build a drawing of *data*, expanding composites down to *depth*.
+
+ The root always expands (if it is a composite); ``depth`` counts further
+ generations below the root's children. Raises ``ValueError`` if ``depth``
+ is negative.
+ """
+ if depth < 0:
+ raise ValueError(f"depth must be >= 0, got {depth}")
+ return _build(data, path="", label=data.recipe.type.value, depth=depth)
+
+
+def _build(
+ data: datastructures.NodeData, path: str, label: str, depth: int
+) -> model.DrawNode:
+ """Build a single node, recursing into children when it is an expanded composite."""
+ inputs, outputs = _ports(data)
+ children: tuple[model.DrawNode, ...] = ()
+ edges: tuple[model.DrawEdge, ...] = ()
+ if isinstance(data, datastructures.CompositeData) and depth >= 0:
+ children = _build_children(data.nodes, path, depth)
+ edges = (
+ _convert_input_edges(data.input_edges)
+ + _convert_sibling_edges(data.edges)
+ + _convert_output_edges(data.output_edges)
+ )
+ return model.DrawNode(
+ path=path,
+ label=label,
+ kind=data.recipe.type,
+ subtitle=style.subtitle_for(data.recipe),
+ inputs=inputs,
+ outputs=outputs,
+ children=children,
+ edges=edges,
+ note=_note(children, edges),
+ )
+
+
+def _note(
+ children: tuple[model.DrawNode, ...], edges: tuple[model.DrawEdge, ...]
+) -> str | None:
+ """A user-facing note when a composite expanded with children but no edges.
+
+ The toy WfMS never records actualized edges for flow-control instances
+ (only ``DagData`` copies edges from its recipe), so a run flow-control node
+ otherwise draws as a set of unconnected boxes.
+ """
+ if children and not edges:
+ return "(no recorded edges)"
+ return None
+
+
+def _ports(
+ data: datastructures.NodeData,
+) -> tuple[tuple[model.DrawPort, ...], tuple[model.DrawPort, ...]]:
+ """Convert a node's data ports to draw ports, annotation hints and all."""
+ inputs = tuple(
+ model.DrawPort(
+ label=label,
+ hint=style.format_annotation(port.annotation),
+ has_default=port.default is not datastructures.NOT_DATA,
+ )
+ for label, port in data.input_ports.items()
+ )
+ outputs = tuple(
+ model.DrawPort(label=label, hint=style.format_annotation(port.annotation))
+ for label, port in data.output_ports.items()
+ )
+ return inputs, outputs
+
+
+def _child_path(parent_path: str, label: str) -> str:
+ """The lexical path of a child node given its parent's path."""
+ return lexical.join(parent_path, label)
+
+
+def _build_children(
+ nodes: Mapping[base_models.Label, datastructures.NodeData],
+ parent_path: str,
+ depth: int,
+) -> tuple[model.DrawNode, ...]:
+ """Build every child of a composite, one generation shallower."""
+ return tuple(
+ _build(
+ child_data, _child_path(parent_path, child_label), child_label, depth - 1
+ )
+ for child_label, child_data in nodes.items()
+ )
+
+
+def _convert_input_edges(edges: edge_models.InputEdges) -> tuple[model.DrawEdge, ...]:
+ """A parent input source, drawn to a real child target."""
+ return tuple(
+ model.DrawEdge(
+ source=model.PortRef("", base_models.IOTypes.INPUTS, source.port),
+ target=model.PortRef(target.node, base_models.IOTypes.INPUTS, target.port),
+ )
+ for target, source in edges.items()
+ )
+
+
+def _convert_sibling_edges(edges: edge_models.Edges) -> tuple[model.DrawEdge, ...]:
+ """A real child source, drawn to a real child target."""
+ return tuple(
+ model.DrawEdge(
+ source=model.PortRef(source.node, base_models.IOTypes.OUTPUTS, source.port),
+ target=model.PortRef(target.node, base_models.IOTypes.INPUTS, target.port),
+ )
+ for target, source in edges.items()
+ )
+
+
+def _convert_output_edges(edges: edge_models.OutputEdges) -> tuple[model.DrawEdge, ...]:
+ """A real child source or a parent-input passthrough, drawn to a parent output."""
+ return tuple(
+ model.DrawEdge(
+ source=_output_edge_source(source),
+ target=model.PortRef("", base_models.IOTypes.OUTPUTS, target.port),
+ )
+ for target, source in edges.items()
+ )
+
+
+def _output_edge_source(
+ source: edge_models.SourceHandle | edge_models.InputSource,
+) -> model.PortRef:
+ """The source endpoint for an output edge: a child output, or a parent passthrough.
+
+ A ``None`` node is what marks a handle as referring to the enclosing node's
+ own IO rather than to a child, so that -- not the handle's class -- is the
+ thing to branch on.
+ """
+ if source.node is None:
+ return model.PortRef("", base_models.IOTypes.INPUTS, source.port)
+ return model.PortRef(source.node, base_models.IOTypes.OUTPUTS, source.port)
diff --git a/src/flowrep/drawing/style.py b/src/flowrep/drawing/style.py
new file mode 100644
index 00000000..2505f2ed
--- /dev/null
+++ b/src/flowrep/drawing/style.py
@@ -0,0 +1,91 @@
+"""
+Visual constants and label formatting for graph drawings.
+
+Deliberately free of any graphviz import so it can be exercised without the
+optional dependency.
+"""
+
+from __future__ import annotations
+
+import typing
+
+from flowrep import base_models
+
+ELLIPSIS = "…"
+
+NODE_PALETTE: dict[base_models.RecipeElementType, tuple[str, str]] = {
+ base_models.RecipeElementType.ATOMIC: ("#dbeafe", "#1d4ed8"),
+ base_models.RecipeElementType.WORKFLOW: ("#dcfce7", "#15803d"),
+ base_models.RecipeElementType.CONSTANT: ("#ccfbf1", "#0f766e"),
+ base_models.RecipeElementType.FOR_EACH: ("#fef3c7", "#b45309"),
+ base_models.RecipeElementType.WHILE: ("#fce7f3", "#be185d"),
+ base_models.RecipeElementType.IF: ("#f3e8ff", "#7e22ce"),
+ base_models.RecipeElementType.TRY: ("#fee2e2", "#b91c1c"),
+}
+
+IO_FILL = "#e5e7eb"
+IO_LINE = "#4b5563"
+CONDITIONAL_LINE = "#6b7280"
+GROUP_LINE = "#9ca3af"
+SUBTITLE_COLOUR = "#374151"
+HINT_COLOUR = "#6b7280"
+
+NODE_LABEL_WRAP = 20
+SUBTITLE_MAX = 28
+PORT_LABEL_MAX = 14
+ANNOTATION_MAX = 16
+
+
+def truncate_right(text: str, limit: int) -> str:
+ """Trim the tail, marking the cut with a trailing ellipsis."""
+ if len(text) <= limit:
+ return text
+ return text[: limit - 1] + ELLIPSIS
+
+
+def truncate_left(text: str, limit: int) -> str:
+ """Trim the head, marking the cut with a leading ellipsis.
+
+ Used for dotted paths, where the qualname at the end is the informative part.
+ """
+ if len(text) <= limit:
+ return text
+ return ELLIPSIS + text[-(limit - 1) :]
+
+
+def wrap_label(text: str, width: int = NODE_LABEL_WRAP) -> list[str]:
+ """Wrap an identifier, preferring breaks just after an underscore."""
+ lines: list[str] = []
+ remaining = text
+ while len(remaining) > width:
+ cut = remaining.rfind("_", 0, width + 1)
+ cut = cut + 1 if cut > 0 else width
+ lines.append(remaining[:cut])
+ remaining = remaining[cut:]
+ lines.append(remaining)
+ return lines
+
+
+def format_annotation(annotation: object | None) -> str | None:
+ """Render a type hint compactly, or ``None`` when there is nothing to show."""
+ if annotation is None:
+ return None
+ if isinstance(annotation, type) and typing.get_origin(annotation) is None:
+ rendered = annotation.__name__
+ else:
+ rendered = str(annotation).replace("typing.", "")
+ return truncate_right(rendered, ANNOTATION_MAX)
+
+
+def subtitle_for(recipe: base_models.NodeRecipe) -> str | None:
+ """The monospace line beneath a node's label, or ``None`` when there is none.
+
+ Dispatches on ``recipe.type`` rather than importing the recipe classes, so
+ this module stays free of any dependence on :mod:`flowrep.prospective`.
+ """
+ if recipe.type is base_models.RecipeElementType.CONSTANT:
+ return truncate_right(repr(getattr(recipe, "constant", None)), SUBTITLE_MAX)
+ fully_qualified_name = getattr(recipe, "fully_qualified_name", None)
+ if fully_qualified_name is None:
+ return None
+ return truncate_left(fully_qualified_name, SUBTITLE_MAX)
diff --git a/src/flowrep/edge_models.py b/src/flowrep/edge_models.py
index d8307fe6..9a49d798 100644
--- a/src/flowrep/edge_models.py
+++ b/src/flowrep/edge_models.py
@@ -4,14 +4,14 @@
import pydantic
-from flowrep import base_models
+from flowrep import base_models, lexical
class HandleModel(pydantic.BaseModel):
model_config = pydantic.ConfigDict(frozen=True)
node: base_models.Label | None
port: base_models.Label
- delimiter: ClassVar[str] = "."
+ delimiter: ClassVar[str] = lexical.DELIMITER
@pydantic.model_serializer
def serialize(self) -> str:
diff --git a/src/flowrep/lexical.py b/src/flowrep/lexical.py
new file mode 100644
index 00000000..ffb60db2
--- /dev/null
+++ b/src/flowrep/lexical.py
@@ -0,0 +1,28 @@
+"""
+Shared helpers for "lexical" paths -- ``"."``-joined node labels, IO type
+segments, and port names.
+
+:cls:`flowrep.base_models.RESERVED_NAMES` forbids nodes and ports from being
+named ``inputs`` or ``outputs``, so a lexical path is unambiguous.
+"""
+
+from __future__ import annotations
+
+from flowrep import base_models
+
+DELIMITER = "."
+
+
+def join(*segments: str) -> str:
+ """Join path segments, ignoring empty ones (e.g. the root's empty path)."""
+ return DELIMITER.join(segment for segment in segments if segment)
+
+
+def split(path: str) -> tuple[str, ...]:
+ """Split a lexical path into its segments; the empty path has none."""
+ return tuple(path.split(DELIMITER)) if path else ()
+
+
+def port_path(node_path: str, io_type: base_models.IOTypes, port: str) -> str:
+ """The lexical path of a port on the node at ``node_path``."""
+ return join(node_path, str(io_type), port)
diff --git a/src/flowrep/retrospective/datastructures.py b/src/flowrep/retrospective/datastructures.py
index 0214976d..1d364b7a 100644
--- a/src/flowrep/retrospective/datastructures.py
+++ b/src/flowrep/retrospective/datastructures.py
@@ -18,10 +18,22 @@
import inspect
import types
from collections.abc import Callable, MutableMapping
-from typing import Any, Generic, Self, TypeVar, get_args, get_origin, get_type_hints
+from typing import (
+ TYPE_CHECKING,
+ Any,
+ Generic,
+ Self,
+ TypeVar,
+ get_args,
+ get_origin,
+ get_type_hints,
+)
from pyiron_snippets import retrieve, singleton
+if TYPE_CHECKING:
+ import graphviz
+
from flowrep import base_models, edge_models
from flowrep.prospective import (
atomic_recipe,
@@ -101,6 +113,29 @@ def view(self, expanded: bool = False):
"""
return viewer.view(self, expanded=expanded)
+ def draw(self, depth: int | None = None) -> graphviz.Digraph:
+ """
+ Draw this data object's topology, ports and labels as a graphviz graph.
+
+ Where :meth:`view` shows the data, this shows the shape. Renders inline in
+ a Jupyter notebook, and also offers ``.render()``, ``.pipe()`` and
+ ``.source``.
+
+ Args:
+ depth: How many generations of nested subgraph to expand below this
+ node's own children. The node itself always expands. Defaults to 0.
+
+ Returns:
+ The drawn graph.
+
+ Raises:
+ ImportAlarmError: If the optional drawing dependency is missing. The
+ message names both the pip and conda install routes.
+ """
+ from flowrep import drawing
+
+ return drawing.draw(self, depth=depth)
+
def _repr_json_(self):
return self.view()._repr_json_()
diff --git a/src/flowrep/retrospective/storage.py b/src/flowrep/retrospective/storage.py
index 14d2155d..3adc16a2 100644
--- a/src/flowrep/retrospective/storage.py
+++ b/src/flowrep/retrospective/storage.py
@@ -12,7 +12,7 @@
from packaging import version
from pyiron_snippets import import_alarm
-from flowrep import base_models
+from flowrep import base_models, lexical
from flowrep.retrospective import datastructures, storage_widget
with import_alarm.ImportAlarm(
@@ -129,7 +129,7 @@ def _collect_lexical_paths(
prefix: str,
paths: list[str],
) -> None:
- for io_type in tuple(base_models.IOTypes):
+ for io_type in base_models.IOTypes:
io_storage = (
_path_to_input_ports(storage_path)
if io_type == base_models.IOTypes.INPUTS
@@ -137,7 +137,9 @@ def _collect_lexical_paths(
)
port_names = bag.open_group(io_storage)
for port in port_names:
- paths.append(f"{prefix}{io_type}.{port}")
+ paths.append(
+ lexical.port_path(prefix.rstrip(lexical.DELIMITER), io_type, port)
+ )
nodes_storage = _path_to_nodes(storage_path)
try:
@@ -145,9 +147,9 @@ def _collect_lexical_paths(
except KeyError:
return
for node in node_names:
- lexical = f"{prefix}{node}"
- paths.append(lexical)
- _collect_lexical_paths(bag, f"{nodes_storage}/{node}", f"{lexical}.", paths)
+ node_path = f"{prefix}{node}"
+ paths.append(node_path)
+ _collect_lexical_paths(bag, f"{nodes_storage}/{node}", f"{node_path}.", paths)
def _path_to_input_ports(path: str) -> str:
diff --git a/tests/unit/drawing/__init__.py b/tests/unit/drawing/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/unit/drawing/test_draw_methods.py b/tests/unit/drawing/test_draw_methods.py
new file mode 100644
index 00000000..00e6d822
--- /dev/null
+++ b/tests/unit/drawing/test_draw_methods.py
@@ -0,0 +1,212 @@
+"""
+The ``.draw()`` convenience methods hung off recipes and data objects.
+
+These are the shortest path to a picture, so they get their own coverage
+separate from the :mod:`flowrep.drawing.interface` callables they delegate to.
+"""
+
+import subprocess
+import sys
+import textwrap
+import unittest
+
+from flowrep import drawing, edge_models, std
+from flowrep.prospective import helper_models, while_recipe, workflow_recipe
+from flowrep.retrospective import datastructures
+
+from flowrep_static import library
+
+try:
+ import graphviz # noqa: F401
+
+ _has_graphviz = True
+except ImportError:
+ _has_graphviz = False
+
+
+def _nested_workflow() -> workflow_recipe.WorkflowRecipe:
+ """A workflow whose child is itself a workflow, so ``depth`` actually bites."""
+ inner = library.simple_workflow.flowrep_recipe
+ return workflow_recipe.WorkflowRecipe(
+ inputs=["p", "q"],
+ outputs=["r"],
+ nodes={"inner": inner},
+ input_edges={
+ edge_models.TargetHandle(node="inner", port="a"): edge_models.InputSource(
+ port="p"
+ ),
+ edge_models.TargetHandle(node="inner", port="b"): edge_models.InputSource(
+ port="q"
+ ),
+ },
+ edges={},
+ output_edges={
+ edge_models.OutputTarget(port="r"): edge_models.SourceHandle(
+ node="inner", port=inner.outputs[0]
+ )
+ },
+ )
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestRecipeDraw(unittest.TestCase):
+ def test_matches_the_module_level_drawer(self):
+ recipe = std.neg.flowrep_recipe
+ self.assertEqual(recipe.draw().source, drawing.draw_prospective(recipe).source)
+
+ def test_default_depth_is_the_prospective_default(self):
+ recipe = _nested_workflow()
+ self.assertEqual(
+ recipe.draw().source, drawing.draw_prospective(recipe, depth=1).source
+ )
+
+ def test_explicit_depth_is_forwarded(self):
+ recipe = _nested_workflow()
+ self.assertEqual(
+ recipe.draw(depth=0).source,
+ drawing.draw_prospective(recipe, depth=0).source,
+ )
+
+ def test_depth_actually_changes_the_drawing(self):
+ recipe = _nested_workflow()
+ self.assertNotEqual(recipe.draw(depth=0).source, recipe.draw(depth=1).source)
+
+ def test_negative_depth_rejected(self):
+ with self.assertRaises(ValueError):
+ std.neg.flowrep_recipe.draw(depth=-1)
+
+ def test_available_on_a_flow_control_recipe(self):
+ """``draw`` lives on the base class, so every recipe type inherits it."""
+ recipe = while_recipe.WhileRecipe(
+ inputs=["x"],
+ outputs=["x"],
+ case=helper_models.ConditionalCase(
+ condition=helper_models.LabeledRecipe(
+ label="cond", recipe=library.is_positive.flowrep_recipe
+ ),
+ body=helper_models.LabeledRecipe(
+ label="body", recipe=library.loop_inc.flowrep_recipe
+ ),
+ ),
+ input_edges={
+ edge_models.TargetHandle(
+ node="cond", port="n"
+ ): edge_models.InputSource(port="x"),
+ edge_models.TargetHandle(
+ node="body", port="x"
+ ): edge_models.InputSource(port="x"),
+ },
+ output_edges={
+ edge_models.OutputTarget(port="x"): edge_models.SourceHandle(
+ node="body", port="y"
+ )
+ },
+ )
+ self.assertIn("cond_i", recipe.draw().source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestDataDraw(unittest.TestCase):
+ def setUp(self):
+ self.data = datastructures.recipe2data(library.simple_workflow.flowrep_recipe)
+
+ def test_matches_the_module_level_drawer(self):
+ self.assertEqual(
+ self.data.draw().source, drawing.draw_retrospective(self.data).source
+ )
+
+ def test_default_depth_is_the_retrospective_default(self):
+ self.assertEqual(
+ self.data.draw().source,
+ drawing.draw_retrospective(self.data, depth=0).source,
+ )
+
+ def test_explicit_depth_is_forwarded(self):
+ self.assertEqual(
+ self.data.draw(depth=1).source,
+ drawing.draw_retrospective(self.data, depth=1).source,
+ )
+
+ def test_negative_depth_rejected(self):
+ with self.assertRaises(ValueError):
+ self.data.draw(depth=-1)
+
+ def test_sits_alongside_view(self):
+ """``view`` shows the data; ``draw`` shows the shape."""
+ self.assertTrue(callable(self.data.view))
+ self.assertTrue(callable(self.data.draw))
+
+
+_WITHOUT_GRAPHVIZ = textwrap.dedent("""
+ import sys
+
+
+ class _BlockGraphviz:
+ def find_spec(self, name, path=None, target=None):
+ if name == "graphviz":
+ raise ImportError("graphviz is blocked for this test")
+ return None
+
+
+ sys.meta_path.insert(0, _BlockGraphviz())
+
+ from flowrep import std
+ from flowrep.retrospective import datastructures
+
+ recipe = std.neg.flowrep_recipe
+ for drawable in (recipe, datastructures.recipe2data(recipe)):
+ try:
+ drawable.draw()
+ except Exception as error:
+ print(f"{type(drawable).__name__}|{type(error).__name__}|{error}")
+ else:
+ print(f"{type(drawable).__name__}|NO_ERROR|")
+ """)
+
+
+class TestMissingDependencyMessage(unittest.TestCase):
+ """
+ Both methods must fail helpfully, not cryptically, in a bare install.
+
+ Run in a subprocess so blocking ``graphviz`` cannot corrupt the interpreter
+ state the rest of the suite shares.
+ """
+
+ @classmethod
+ def setUpClass(cls):
+ result = subprocess.run(
+ [sys.executable, "-c", _WITHOUT_GRAPHVIZ],
+ capture_output=True,
+ text=True,
+ check=True,
+ )
+ cls.lines = [
+ line.split("|", 2) for line in result.stdout.strip().splitlines() if line
+ ]
+
+ def test_both_drawables_raise(self):
+ self.assertEqual(len(self.lines), 2)
+ for name, error, _ in self.lines:
+ with self.subTest(drawable=name):
+ self.assertEqual(error, "ImportAlarmError")
+
+ def test_message_names_the_package(self):
+ for name, _, message in self.lines:
+ with self.subTest(drawable=name):
+ self.assertIn("graphviz", message)
+
+ def test_message_names_the_pip_route_and_its_binary_caveat(self):
+ for name, _, message in self.lines:
+ with self.subTest(drawable=name):
+ self.assertIn("pip install flowrep[drawing]", message)
+ self.assertIn("dot", message)
+
+ def test_message_names_the_conda_route(self):
+ for name, _, message in self.lines:
+ with self.subTest(drawable=name):
+ self.assertIn("conda install", message)
+ self.assertIn("python-graphviz", message)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/unit/drawing/test_interface.py b/tests/unit/drawing/test_interface.py
new file mode 100644
index 00000000..268c94d2
--- /dev/null
+++ b/tests/unit/drawing/test_interface.py
@@ -0,0 +1,151 @@
+import subprocess
+import sys
+import unittest
+
+import flowrep
+from flowrep import edge_models, std
+from flowrep.prospective import workflow_recipe
+from flowrep.retrospective import datastructures
+
+from flowrep_static import library
+
+try:
+ import graphviz # noqa: F401
+
+ _has_graphviz = True
+except ImportError:
+ _has_graphviz = False
+
+
+def _nested_workflow() -> workflow_recipe.WorkflowRecipe:
+ """A workflow whose child is itself a workflow, so depth actually bites.
+
+ ``simple_workflow`` alone will not do: its only child is atomic, and an
+ atomic node is a leaf at every depth, so depth 0 and depth 1 would render
+ identically.
+ """
+ inner = library.simple_workflow.flowrep_recipe
+ return workflow_recipe.WorkflowRecipe(
+ inputs=["p", "q"],
+ outputs=["r"],
+ nodes={"inner": inner},
+ input_edges={
+ edge_models.TargetHandle(node="inner", port="a"): edge_models.InputSource(
+ port="p"
+ ),
+ edge_models.TargetHandle(node="inner", port="b"): edge_models.InputSource(
+ port="q"
+ ),
+ },
+ edges={},
+ output_edges={
+ edge_models.OutputTarget(port="r"): edge_models.SourceHandle(
+ node="inner", port=inner.outputs[0]
+ )
+ },
+ )
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestDispatch(unittest.TestCase):
+ def test_recipe_dispatches_prospective(self):
+ self.assertIn("neg", flowrep.draw(std.neg.flowrep_recipe).source)
+
+ def test_data_dispatches_retrospective(self):
+ data = datastructures.recipe2data(library.decrement.flowrep_recipe)
+ self.assertIn("decrement", flowrep.draw(data).source)
+
+ def test_unknown_type_raises(self):
+ with self.assertRaises(TypeError):
+ flowrep.draw("not a graph")
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestDepthDefaults(unittest.TestCase):
+ def setUp(self):
+ self.recipe = _nested_workflow()
+
+ def test_prospective_default_is_one(self):
+ self.assertEqual(
+ flowrep.drawing.draw_prospective(self.recipe).source,
+ flowrep.drawing.draw_prospective(self.recipe, depth=1).source,
+ )
+
+ def test_explicit_depth_overrides(self):
+ shallow = flowrep.drawing.draw_prospective(self.recipe, depth=0).source
+ deep = flowrep.drawing.draw_prospective(self.recipe, depth=1).source
+ self.assertNotEqual(shallow, deep)
+
+ def test_retrospective_default_is_zero(self):
+ data = datastructures.recipe2data(self.recipe)
+ self.assertEqual(
+ flowrep.drawing.draw_retrospective(data).source,
+ flowrep.drawing.draw_retrospective(data, depth=0).source,
+ )
+
+ def test_dispatcher_forwards_explicit_depth(self):
+ self.assertEqual(
+ flowrep.draw(self.recipe, depth=0).source,
+ flowrep.drawing.draw_prospective(self.recipe, depth=0).source,
+ )
+
+
+class TestExposure(unittest.TestCase):
+ def test_draw_is_top_level(self):
+ self.assertTrue(callable(flowrep.draw))
+
+ def test_draw_is_in_tools(self):
+ self.assertIs(flowrep.tools.draw, flowrep.draw)
+
+ def test_specific_drawers_stay_in_the_subpackage(self):
+ self.assertFalse(hasattr(flowrep, "draw_prospective"))
+ self.assertTrue(callable(flowrep.drawing.draw_prospective))
+ self.assertTrue(callable(flowrep.drawing.draw_retrospective))
+
+
+class TestImportSafetyWithoutGraphviz(unittest.TestCase):
+ """``import flowrep`` must succeed even without graphviz installed.
+
+ ``flowrep/__init__.py`` transitively imports ``flowrep.drawing.render``,
+ which imports graphviz. The ``ImportAlarm(..., raise_exception=True)``
+ guard there is meant to swallow the ``ImportError`` at import time and
+ re-raise only when a drawing callable is actually invoked. This is run in
+ a subprocess, with a meta path finder blocking graphviz, so it exercises a
+ genuinely fresh import rather than relying on already-imported modules.
+ """
+
+ _SCRIPT = """
+import sys
+import importlib.abc
+
+
+class _BlockGraphviz(importlib.abc.MetaPathFinder):
+ def find_spec(self, name, path, target=None):
+ if name == "graphviz" or name.startswith("graphviz."):
+ raise ImportError("graphviz blocked for test")
+ return None
+
+
+sys.meta_path.insert(0, _BlockGraphviz())
+
+import flowrep
+from flowrep import std
+
+assert callable(flowrep.draw), "flowrep.draw must be importable without graphviz"
+
+try:
+ flowrep.draw(std.neg.flowrep_recipe)
+except ImportError:
+ print("RAISED")
+else:
+ print("DID NOT RAISE")
+"""
+
+ def test_import_succeeds_and_draw_raises_without_graphviz(self):
+ result = subprocess.run(
+ [sys.executable, "-c", self._SCRIPT],
+ capture_output=True,
+ text=True,
+ )
+ self.assertEqual(result.returncode, 0, msg=result.stderr)
+ self.assertEqual(result.stdout.strip(), "RAISED")
diff --git a/tests/unit/drawing/test_model.py b/tests/unit/drawing/test_model.py
new file mode 100644
index 00000000..709d6649
--- /dev/null
+++ b/tests/unit/drawing/test_model.py
@@ -0,0 +1,83 @@
+import dataclasses
+import unittest
+
+from flowrep import base_models
+from flowrep.drawing import model
+
+
+def _port(label: str) -> model.DrawPort:
+ return model.DrawPort(label=label, hint=None, has_default=False, badge=None)
+
+
+def _leaf(path: str, label: str) -> model.DrawNode:
+ return model.DrawNode(
+ path=path,
+ label=label,
+ kind=base_models.RecipeElementType.ATOMIC,
+ subtitle=None,
+ inputs=(_port("x"),),
+ outputs=(_port("y"),),
+ children=(),
+ edges=(),
+ )
+
+
+class TestDrawNode(unittest.TestCase):
+ def test_is_frozen(self):
+ node = _leaf("a", "a")
+ with self.assertRaises(dataclasses.FrozenInstanceError):
+ node.path = "b" # type: ignore[misc]
+
+ def test_leaf_has_no_children(self):
+ self.assertTrue(_leaf("a", "a").is_leaf)
+
+ def test_composite_is_not_leaf(self):
+ parent = model.DrawNode(
+ path="",
+ label="wf",
+ kind=base_models.RecipeElementType.WORKFLOW,
+ subtitle=None,
+ inputs=(),
+ outputs=(),
+ children=(_leaf("a", "a"),),
+ edges=(),
+ )
+ self.assertFalse(parent.is_leaf)
+
+ def test_walk_is_self_then_descendants(self):
+ grandchild = _leaf("a.b", "b")
+ child = dataclasses.replace(_leaf("a", "a"), children=(grandchild,))
+ root = dataclasses.replace(_leaf("", "root"), children=(child,))
+ self.assertEqual([n.path for n in root.walk()], ["", "a", "a.b"])
+
+ def test_optional_fields_default(self):
+ node = _leaf("a", "a")
+ self.assertEqual(node.groups, ())
+ self.assertIsNone(node.note)
+
+
+class TestDrawEdge(unittest.TestCase):
+ def test_defaults_to_unconditional(self):
+ edge = model.DrawEdge(
+ source=model.PortRef("a", base_models.IOTypes.OUTPUTS, "y"),
+ target=model.PortRef("b", base_models.IOTypes.INPUTS, "x"),
+ )
+ self.assertFalse(edge.conditional)
+
+ def test_is_hashable(self):
+ """Edges land in sets during cycle detection."""
+ edge = model.DrawEdge(
+ source=model.PortRef("a", base_models.IOTypes.OUTPUTS, "y"),
+ target=model.PortRef("b", base_models.IOTypes.INPUTS, "x"),
+ )
+ self.assertEqual(len({edge, edge}), 1)
+
+
+class TestPortRef(unittest.TestCase):
+ def test_lexical_path(self):
+ ref = model.PortRef("sub.body_0", base_models.IOTypes.INPUTS, "x")
+ self.assertEqual(ref.lexical_path, "sub.body_0.inputs.x")
+
+ def test_empty_node_path_means_enclosing_node(self):
+ ref = model.PortRef("", base_models.IOTypes.OUTPUTS, "y")
+ self.assertEqual(ref.lexical_path, "outputs.y")
diff --git a/tests/unit/drawing/test_prospective.py b/tests/unit/drawing/test_prospective.py
new file mode 100644
index 00000000..1212bb3f
--- /dev/null
+++ b/tests/unit/drawing/test_prospective.py
@@ -0,0 +1,198 @@
+import unittest
+from typing import Literal
+
+import pydantic
+
+from flowrep import base_models, edge_models, std
+from flowrep.drawing import prospective
+from flowrep.prospective import (
+ constant_recipe,
+ workflow_recipe,
+)
+
+from flowrep_static import library
+
+
+def _by_path(graph, path):
+ for node in graph.walk():
+ if node.path == path:
+ return node
+ raise AssertionError(f"no node at {path!r}; have {[n.path for n in graph.walk()]}")
+
+
+class _UnrecognizedRecipe(base_models.NodeRecipe):
+ """A well-formed recipe of a type the drawing builder has no branch for."""
+
+ type: Literal[base_models.RecipeElementType.ATOMIC] = pydantic.Field(
+ default=base_models.RecipeElementType.ATOMIC, frozen=True
+ )
+
+ def __call__(self, *args, **kwargs):
+ raise NotImplementedError()
+
+
+class TestDepthValidation(unittest.TestCase):
+ def test_negative_depth_rejected(self):
+ with self.assertRaises(ValueError):
+ prospective.build(std.neg.flowrep_recipe, depth=-1)
+
+
+class TestAtomic(unittest.TestCase):
+ def setUp(self):
+ self.graph = prospective.build(std.neg.flowrep_recipe)
+
+ def test_root_path_is_empty(self):
+ self.assertEqual(self.graph.path, "")
+
+ def test_atomic_is_always_a_leaf(self):
+ self.assertTrue(self.graph.is_leaf)
+
+ def test_kind(self):
+ self.assertEqual(self.graph.kind, base_models.RecipeElementType.ATOMIC)
+
+ def test_ports(self):
+ self.assertEqual([p.label for p in self.graph.inputs], ["a"])
+ self.assertEqual([p.label for p in self.graph.outputs], ["negative"])
+
+ def test_subtitle_is_left_truncated_qualified_name(self):
+ self.assertTrue(self.graph.subtitle.endswith("neg"))
+
+ def test_no_prospective_hints(self):
+ """Recipes carry no annotations; we do not import references to find any."""
+ self.assertTrue(all(p.hint is None for p in self.graph.inputs))
+
+
+class TestDefaultsAreItalicised(unittest.TestCase):
+ def test_input_with_default_flagged(self):
+ graph = prospective.build(library.increment.flowrep_recipe)
+ flags = {p.label: p.has_default for p in graph.inputs}
+ self.assertFalse(flags["x"])
+ self.assertTrue(flags["step"])
+
+
+class TestConstant(unittest.TestCase):
+ def test_subtitle_is_the_repr(self):
+ graph = prospective.build(constant_recipe.ConstantRecipe(constant=42))
+ self.assertEqual(graph.subtitle, "42")
+
+ def test_single_output_no_inputs(self):
+ graph = prospective.build(constant_recipe.ConstantRecipe(constant=42))
+ self.assertEqual(graph.inputs, ())
+ self.assertEqual([p.label for p in graph.outputs], ["constant"])
+
+
+class TestWorkflow(unittest.TestCase):
+ def setUp(self):
+ self.recipe = library.simple_workflow.flowrep_recipe
+ self.graph = prospective.build(self.recipe, depth=0)
+
+ def test_children_present_at_depth_zero(self):
+ self.assertEqual(
+ sorted(c.path for c in self.graph.children), sorted(self.recipe.nodes)
+ )
+
+ def test_children_are_leaves_at_depth_zero(self):
+ self.assertTrue(all(c.is_leaf for c in self.graph.children))
+
+ def test_edge_count_matches_recipe(self):
+ expected = (
+ len(self.recipe.input_edges)
+ + len(self.recipe.edges)
+ + len(self.recipe.output_edges)
+ )
+ self.assertEqual(len(self.graph.edges), expected)
+
+ def test_input_edge_sources_the_parent(self):
+ """A parent-input source is encoded as the empty node path."""
+ target, source = next(iter(self.recipe.input_edges.items()))
+ match = [
+ e
+ for e in self.graph.edges
+ if e.target.node_path == target.node and e.target.port == target.port
+ ]
+ self.assertEqual(len(match), 1)
+ self.assertEqual(match[0].source.node_path, "")
+ self.assertEqual(match[0].source.port, source.port)
+
+ def test_nothing_is_conditional(self):
+ self.assertTrue(all(not e.conditional for e in self.graph.edges))
+
+
+class TestNestedDepth(unittest.TestCase):
+ """A workflow whose child is itself a workflow."""
+
+ def setUp(self):
+ inner = library.simple_workflow.flowrep_recipe
+ self.recipe = workflow_recipe.WorkflowRecipe(
+ inputs=["p", "q"],
+ outputs=["r"],
+ nodes={"inner": inner},
+ input_edges={
+ edge_models.TargetHandle(
+ node="inner", port="a"
+ ): edge_models.InputSource(port="p"),
+ edge_models.TargetHandle(
+ node="inner", port="b"
+ ): edge_models.InputSource(port="q"),
+ },
+ edges={},
+ output_edges={
+ edge_models.OutputTarget(port="r"): edge_models.SourceHandle(
+ node="inner", port=inner.outputs[0]
+ )
+ },
+ )
+
+ def test_depth_zero_leaves_the_child_closed(self):
+ graph = prospective.build(self.recipe, depth=0)
+ self.assertTrue(_by_path(graph, "inner").is_leaf)
+
+ def test_depth_one_opens_the_child(self):
+ graph = prospective.build(self.recipe, depth=1)
+ self.assertFalse(_by_path(graph, "inner").is_leaf)
+
+ def test_grandchild_paths_are_lexical(self):
+ graph = prospective.build(self.recipe, depth=1)
+ inner = _by_path(graph, "inner")
+ for child in inner.children:
+ with self.subTest(path=child.path):
+ self.assertTrue(child.path.startswith("inner."))
+
+ def test_depth_one_grandchildren_are_leaves(self):
+ graph = prospective.build(self.recipe, depth=1)
+ self.assertTrue(all(c.is_leaf for c in _by_path(graph, "inner").children))
+
+
+class TestOutputPassthrough(unittest.TestCase):
+ """An output_edges entry sourced from the parent's own input, not a child."""
+
+ def test_passthrough_source_is_the_parent_input(self):
+ recipe = workflow_recipe.WorkflowRecipe(
+ inputs=["a"],
+ outputs=["passthrough"],
+ nodes={},
+ input_edges={},
+ edges={},
+ output_edges={
+ edge_models.OutputTarget(port="passthrough"): edge_models.InputSource(
+ port="a"
+ ),
+ },
+ )
+ graph = prospective.build(recipe, depth=0)
+ self.assertEqual(len(graph.edges), 1)
+ edge = graph.edges[0]
+ self.assertEqual(edge.source.node_path, "")
+ self.assertEqual(edge.source.io_type, base_models.IOTypes.INPUTS)
+ self.assertEqual(edge.source.port, "a")
+ self.assertEqual(edge.target.node_path, "")
+ self.assertEqual(edge.target.io_type, base_models.IOTypes.OUTPUTS)
+ self.assertEqual(edge.target.port, "passthrough")
+
+
+class TestUnrecognizedRecipe(unittest.TestCase):
+ def test_unrecognized_recipe_subclass_raises(self):
+ """A NodeRecipe the dispatch has no branch for still falls through."""
+ with self.assertRaises(TypeError) as ctx:
+ prospective.build(_UnrecognizedRecipe(inputs=[], outputs=[]))
+ self.assertIn("Unrecognized recipe type", str(ctx.exception))
diff --git a/tests/unit/drawing/test_prospective_flow_control.py b/tests/unit/drawing/test_prospective_flow_control.py
new file mode 100644
index 00000000..57c6c398
--- /dev/null
+++ b/tests/unit/drawing/test_prospective_flow_control.py
@@ -0,0 +1,289 @@
+import unittest
+
+from pyiron_snippets import versions
+
+from flowrep import edge_models, std
+from flowrep.drawing import prospective
+from flowrep.prospective import (
+ for_recipe,
+ helper_models,
+ if_recipe,
+ try_recipe,
+ while_recipe,
+)
+
+from flowrep_static import library
+
+
+def _by_path(graph, path):
+ for node in graph.walk():
+ if node.path == path:
+ return node
+ raise AssertionError(f"no node at {path!r}; have {[n.path for n in graph.walk()]}")
+
+
+def _make_for() -> for_recipe.ForEachRecipe:
+ return for_recipe.ForEachRecipe(
+ inputs=["xs"],
+ outputs=["ys", "used"],
+ body_node=helper_models.LabeledRecipe(
+ label="body", recipe=std.neg.flowrep_recipe
+ ),
+ input_edges={
+ edge_models.TargetHandle(node="body", port="a"): edge_models.InputSource(
+ port="xs"
+ ),
+ },
+ output_edges={
+ edge_models.OutputTarget(port="ys"): edge_models.SourceHandle(
+ node="body", port="negative"
+ ),
+ edge_models.OutputTarget(port="used"): edge_models.InputSource(port="xs"),
+ },
+ nested_ports=["a"],
+ )
+
+
+def _make_while() -> while_recipe.WhileRecipe:
+ return while_recipe.WhileRecipe(
+ inputs=["x"],
+ outputs=["x"],
+ case=helper_models.ConditionalCase(
+ condition=helper_models.LabeledRecipe(
+ label="cond", recipe=library.is_positive.flowrep_recipe
+ ),
+ body=helper_models.LabeledRecipe(
+ label="body", recipe=library.loop_inc.flowrep_recipe
+ ),
+ ),
+ input_edges={
+ edge_models.TargetHandle(node="cond", port="n"): edge_models.InputSource(
+ port="x"
+ ),
+ edge_models.TargetHandle(node="body", port="x"): edge_models.InputSource(
+ port="x"
+ ),
+ },
+ output_edges={
+ edge_models.OutputTarget(port="x"): edge_models.SourceHandle(
+ node="body", port="y"
+ )
+ },
+ )
+
+
+def _make_if() -> if_recipe.IfRecipe:
+ return if_recipe.IfRecipe(
+ inputs=["n"],
+ outputs=["out"],
+ cases=[
+ helper_models.ConditionalCase(
+ condition=helper_models.LabeledRecipe(
+ label="cond0", recipe=library.is_positive.flowrep_recipe
+ ),
+ body=helper_models.LabeledRecipe(
+ label="body0", recipe=library.increment.flowrep_recipe
+ ),
+ )
+ ],
+ else_case=helper_models.LabeledRecipe(
+ label="otherwise", recipe=library.decrement.flowrep_recipe
+ ),
+ input_edges={
+ edge_models.TargetHandle(node="cond0", port="n"): edge_models.InputSource(
+ port="n"
+ ),
+ edge_models.TargetHandle(node="body0", port="x"): edge_models.InputSource(
+ port="n"
+ ),
+ edge_models.TargetHandle(
+ node="otherwise", port="x"
+ ): edge_models.InputSource(port="n"),
+ },
+ prospective_output_edges={
+ edge_models.OutputTarget(port="out"): [
+ edge_models.SourceHandle(node="body0", port="output_0"),
+ edge_models.SourceHandle(node="otherwise", port="output_0"),
+ ]
+ },
+ )
+
+
+def _make_try() -> try_recipe.TryRecipe:
+ return try_recipe.TryRecipe(
+ inputs=["x", "y"],
+ outputs=["out"],
+ try_node=helper_models.LabeledRecipe(
+ label="attempt", recipe=library.raises_custom.flowrep_recipe
+ ),
+ exception_cases=[
+ helper_models.ExceptionCase(
+ exceptions=[versions.VersionInfo.of(library.MyCustomException)],
+ body=helper_models.LabeledRecipe(
+ label="handler", recipe=library.combine.flowrep_recipe
+ ),
+ )
+ ],
+ input_edges={
+ edge_models.TargetHandle(node="attempt", port="x"): edge_models.InputSource(
+ port="x"
+ ),
+ edge_models.TargetHandle(node="attempt", port="y"): edge_models.InputSource(
+ port="y"
+ ),
+ edge_models.TargetHandle(node="handler", port="a"): edge_models.InputSource(
+ port="x"
+ ),
+ edge_models.TargetHandle(node="handler", port="b"): edge_models.InputSource(
+ port="y"
+ ),
+ },
+ prospective_output_edges={
+ edge_models.OutputTarget(port="out"): [
+ edge_models.SourceHandle(
+ node="attempt", port=library.raises_custom.flowrep_recipe.outputs[0]
+ ),
+ edge_models.SourceHandle(
+ node="handler", port=library.combine.flowrep_recipe.outputs[0]
+ ),
+ ]
+ },
+ )
+
+
+class TestForEach(unittest.TestCase):
+ def setUp(self):
+ self.graph = prospective.build(_make_for(), depth=0)
+
+ def test_body_expands_once(self):
+ self.assertEqual([c.path for c in self.graph.children], ["body"])
+
+ def test_body_display_label_signals_multiplicity(self):
+ self.assertEqual(self.graph.children[0].label, "body_n")
+
+ def test_path_keeps_the_real_label(self):
+ """Identity stays honest even though the display label is decorated."""
+ self.assertEqual(self.graph.children[0].path, "body")
+
+ def test_nested_port_badged(self):
+ badges = {p.label: p.badge for p in self.graph.children[0].inputs}
+ self.assertEqual(badges["a"], "nested")
+
+ def test_transferred_output_links_input_to_output(self):
+ matches = [
+ e
+ for e in self.graph.edges
+ if e.source.node_path == ""
+ and e.target.node_path == ""
+ and e.target.port == "used"
+ ]
+ self.assertEqual(len(matches), 1)
+ self.assertEqual(matches[0].source.port, "xs")
+
+
+class TestForEachZipped(unittest.TestCase):
+ def test_zipped_port_badged(self):
+ recipe = _make_for().model_copy(
+ update={"nested_ports": [], "zipped_ports": ["a"]}
+ )
+ graph = prospective.build(recipe, depth=0)
+ badges = {p.label: p.badge for p in graph.children[0].inputs}
+ self.assertEqual(badges["a"], "zipped")
+
+
+class TestWhile(unittest.TestCase):
+ def setUp(self):
+ self.graph = prospective.build(_make_while(), depth=0)
+
+ def test_condition_and_body_both_present(self):
+ self.assertEqual(sorted(c.path for c in self.graph.children), ["body", "cond"])
+
+ def test_iteration_suffix_on_display_labels(self):
+ self.assertEqual(
+ sorted(c.label for c in self.graph.children), ["body_i", "cond_i"]
+ )
+
+ def test_condition_output_badged(self):
+ cond = _by_path(self.graph, "cond")
+ self.assertEqual([p.badge for p in cond.outputs], ["test"])
+
+ def test_inferred_loop_back_edges_present(self):
+ back = [
+ e
+ for e in self.graph.edges
+ if e.source.node_path == "body" and e.target.node_path in ("body", "cond")
+ ]
+ self.assertEqual(len(back), 2)
+
+ def test_fallback_edge_is_conditional(self):
+ fallback = [
+ e
+ for e in self.graph.edges
+ if e.source.node_path == "" and e.target.node_path == "" and e.conditional
+ ]
+ self.assertEqual(len(fallback), 1)
+ self.assertEqual(fallback[0].source.port, "x")
+ self.assertEqual(fallback[0].target.port, "x")
+
+
+class TestIf(unittest.TestCase):
+ def setUp(self):
+ self.graph = prospective.build(_make_if(), depth=0)
+
+ def test_all_branch_nodes_present(self):
+ self.assertEqual(
+ sorted(c.path for c in self.graph.children),
+ ["body0", "cond0", "otherwise"],
+ )
+
+ def test_groups_pair_condition_with_body(self):
+ groups = {g.label: g.members for g in self.graph.groups}
+ self.assertEqual(sorted(groups["case 0"]), ["body0", "cond0"])
+
+ def test_else_group(self):
+ groups = {g.label: g.members for g in self.graph.groups}
+ self.assertEqual(groups["else"], ("otherwise",))
+
+ def test_group_order_follows_declaration(self):
+ self.assertEqual([g.label for g in self.graph.groups], ["case 0", "else"])
+
+ def test_candidate_output_edges_are_conditional(self):
+ to_out = [e for e in self.graph.edges if e.target.node_path == ""]
+ self.assertEqual(len(to_out), 2)
+ self.assertTrue(all(e.conditional for e in to_out))
+
+ def test_condition_output_badged(self):
+ self.assertEqual(
+ [p.badge for p in _by_path(self.graph, "cond0").outputs], ["test"]
+ )
+
+
+class TestTry(unittest.TestCase):
+ def setUp(self):
+ self.graph = prospective.build(_make_try(), depth=0)
+
+ def test_try_and_handler_present(self):
+ self.assertEqual(
+ sorted(c.path for c in self.graph.children), ["attempt", "handler"]
+ )
+
+ def test_try_group(self):
+ groups = {g.label: g.members for g in self.graph.groups}
+ self.assertEqual(groups["try"], ("attempt",))
+
+ def test_exception_group_names_the_exception(self):
+ labels = [g.label for g in self.graph.groups]
+ self.assertTrue(
+ any(label.startswith("except ") for label in labels), msg=str(labels)
+ )
+
+ def test_candidate_output_edges_are_conditional(self):
+ to_out = [e for e in self.graph.edges if e.target.node_path == ""]
+ self.assertEqual(len(to_out), 2)
+ self.assertTrue(all(e.conditional for e in to_out))
+
+
+class TestFlowControlDepth(unittest.TestCase):
+ def test_for_body_is_a_leaf_when_atomic(self):
+ graph = prospective.build(_make_for(), depth=1)
+ self.assertTrue(_by_path(graph, "body").is_leaf)
diff --git a/tests/unit/drawing/test_render.py b/tests/unit/drawing/test_render.py
new file mode 100644
index 00000000..b839e8ca
--- /dev/null
+++ b/tests/unit/drawing/test_render.py
@@ -0,0 +1,305 @@
+import shutil
+import unittest
+
+from flowrep import base_models, edge_models, std, wfms
+from flowrep.drawing import model, prospective, render, retrospective, style
+from flowrep.prospective import (
+ constant_recipe,
+ for_recipe,
+ helper_models,
+ if_recipe,
+ while_recipe,
+ workflow_recipe,
+)
+
+from flowrep_static import library
+
+try:
+ import graphviz # noqa: F401
+
+ _has_graphviz = True
+except ImportError:
+ _has_graphviz = False
+
+
+def _while_recipe():
+ return while_recipe.WhileRecipe(
+ inputs=["x"],
+ outputs=["x"],
+ case=helper_models.ConditionalCase(
+ condition=helper_models.LabeledRecipe(
+ label="cond", recipe=library.is_positive.flowrep_recipe
+ ),
+ body=helper_models.LabeledRecipe(
+ label="body", recipe=library.loop_inc.flowrep_recipe
+ ),
+ ),
+ input_edges={
+ edge_models.TargetHandle(node="cond", port="n"): edge_models.InputSource(
+ port="x"
+ ),
+ edge_models.TargetHandle(node="body", port="x"): edge_models.InputSource(
+ port="x"
+ ),
+ },
+ output_edges={
+ edge_models.OutputTarget(port="x"): edge_models.SourceHandle(
+ node="body", port="y"
+ )
+ },
+ )
+
+
+def _for_recipe():
+ return for_recipe.ForEachRecipe(
+ inputs=["xs"],
+ outputs=["ys"],
+ body_node=helper_models.LabeledRecipe(
+ label="body", recipe=std.neg.flowrep_recipe
+ ),
+ input_edges={
+ edge_models.TargetHandle(node="body", port="a"): edge_models.InputSource(
+ port="xs"
+ )
+ },
+ output_edges={
+ edge_models.OutputTarget(port="ys"): edge_models.SourceHandle(
+ node="body", port="negative"
+ )
+ },
+ nested_ports=["a"],
+ )
+
+
+def _if_recipe():
+ return if_recipe.IfRecipe(
+ inputs=["n"],
+ outputs=["out"],
+ cases=[
+ helper_models.ConditionalCase(
+ condition=helper_models.LabeledRecipe(
+ label="cond0", recipe=library.is_positive.flowrep_recipe
+ ),
+ body=helper_models.LabeledRecipe(
+ label="body0", recipe=library.increment.flowrep_recipe
+ ),
+ )
+ ],
+ else_case=helper_models.LabeledRecipe(
+ label="otherwise", recipe=library.decrement.flowrep_recipe
+ ),
+ input_edges={
+ edge_models.TargetHandle(node="cond0", port="n"): edge_models.InputSource(
+ port="n"
+ ),
+ edge_models.TargetHandle(node="body0", port="x"): edge_models.InputSource(
+ port="n"
+ ),
+ edge_models.TargetHandle(
+ node="otherwise", port="x"
+ ): edge_models.InputSource(port="n"),
+ },
+ prospective_output_edges={
+ edge_models.OutputTarget(port="out"): [
+ edge_models.SourceHandle(node="body0", port="output_0"),
+ edge_models.SourceHandle(node="otherwise", port="output_0"),
+ ]
+ },
+ )
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestLeafRendering(unittest.TestCase):
+ def setUp(self):
+ self.source = render.render(prospective.build(std.neg.flowrep_recipe)).source
+
+ def test_is_left_to_right(self):
+ self.assertIn("rankdir=LR", self.source)
+
+ def test_node_label_present(self):
+ self.assertIn("neg", self.source)
+
+ def test_ports_are_addressable(self):
+ self.assertIn('PORT="i_a"', self.source)
+ self.assertIn('PORT="o_negative"', self.source)
+
+ def test_atomic_fill_colour_used(self):
+ fill, _ = style.NODE_PALETTE[base_models.RecipeElementType.ATOMIC]
+ self.assertIn(fill, self.source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestCompositeRendering(unittest.TestCase):
+ def setUp(self):
+ graph = prospective.build(library.simple_workflow.flowrep_recipe, depth=0)
+ self.source = render.render(graph).source
+
+ def test_root_becomes_a_cluster(self):
+ self.assertIn("subgraph cluster_", self.source)
+
+ def test_child_ids_appear_in_source(self):
+ """A single-segment lexical path is a valid DOT identifier and Graphviz
+ does not quote it unnecessarily; quoting is exercised separately for a
+ genuinely nested (dotted) path."""
+ for child in prospective.build(
+ library.simple_workflow.flowrep_recipe, depth=0
+ ).children:
+ with self.subTest(path=child.path):
+ self.assertIn(child.path, self.source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestNestedIdsAreQuoted(unittest.TestCase):
+ def test_dotted_child_path_is_quoted(self):
+ """A dotted lexical path is not a bare DOT identifier, so Graphviz quotes it."""
+ inner = library.simple_workflow.flowrep_recipe
+ outer = workflow_recipe.WorkflowRecipe(
+ inputs=["p", "q"],
+ outputs=["r"],
+ nodes={"inner": inner},
+ input_edges={
+ edge_models.TargetHandle(
+ node="inner", port="a"
+ ): edge_models.InputSource(port="p"),
+ edge_models.TargetHandle(
+ node="inner", port="b"
+ ): edge_models.InputSource(port="q"),
+ },
+ edges={},
+ output_edges={
+ edge_models.OutputTarget(port="r"): edge_models.SourceHandle(
+ node="inner", port=inner.outputs[0]
+ )
+ },
+ )
+ source = render.render(prospective.build(outer, depth=1)).source
+ self.assertIn('"inner.typed_add_0"', source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestCycleHandling(unittest.TestCase):
+ def test_back_edges_are_unconstrained(self):
+ """While-loop feedback would otherwise destroy the rankdir=LR layout."""
+ source = render.render(prospective.build(_while_recipe(), depth=0)).source
+ self.assertIn("constraint=false", source)
+
+ def test_acyclic_composite_has_no_unconstrained_edges(self):
+ """A plain DAG must not be mistaken for a cycle just because its own
+ input and its own output share the empty ``node_path`` placeholder."""
+ source = render.render(
+ prospective.build(library.simple_workflow.flowrep_recipe, depth=0)
+ ).source
+ self.assertNotIn("constraint=false", source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestConditionalEdges(unittest.TestCase):
+ def test_conditional_edges_dashed(self):
+ source = render.render(prospective.build(_while_recipe(), depth=0)).source
+ self.assertIn("style=dashed", source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestGroupOrdering(unittest.TestCase):
+ def test_groups_emit_in_reverse_declaration_order(self):
+ """Under rankdir=LR, Graphviz stacks same-rank clusters bottom-up in
+ declaration order, so the first-declared group ("case 0") must be the
+ *last* one written to source for it to land on top."""
+ source = render.render(prospective.build(_if_recipe(), depth=0)).source
+ self.assertLess(source.index("else"), source.index("case 0"))
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestEscaping(unittest.TestCase):
+ def test_special_characters_in_subtitle_are_escaped(self):
+ graph = prospective.build(constant_recipe.ConstantRecipe(constant=""))
+ source = render.render(graph).source
+ self.assertNotIn("", source)
+ self.assertIn("<a & b>", source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestPortPaddingBranches(unittest.TestCase):
+ def test_more_inputs_than_outputs_pads_output_column(self):
+ source = render.render(prospective.build(library.combine.flowrep_recipe)).source
+ self.assertIn(" | ", source)
+
+ def test_more_outputs_than_inputs_pads_input_column(self):
+ source = render.render(
+ prospective.build(constant_recipe.ConstantRecipe(constant=1))
+ ).source
+ self.assertIn(" | ", source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestEmptyIoComposite(unittest.TestCase):
+ def test_composite_with_no_own_io_draws_no_io_boxes(self):
+ recipe = workflow_recipe.WorkflowRecipe(
+ inputs=[],
+ outputs=[],
+ nodes={"c": constant_recipe.ConstantRecipe(constant=1)},
+ input_edges={},
+ edges={},
+ output_edges={},
+ )
+ source = render.render(prospective.build(recipe, depth=0)).source
+ self.assertNotIn("inputs.", source)
+ self.assertNotIn("outputs.", source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestMultipleOwnOutputsRankSame(unittest.TestCase):
+ def test_two_own_outputs_are_rank_aligned(self):
+ source = render.render(
+ prospective.build(library.autoencoder.flowrep_recipe, depth=0)
+ ).source
+ self.assertIn("rank=same", source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestBadgePreferredOverHint(unittest.TestCase):
+ def test_badge_wins_when_both_set(self):
+ node = model.DrawNode(
+ path="",
+ label="x",
+ kind=base_models.RecipeElementType.ATOMIC,
+ subtitle=None,
+ inputs=(model.DrawPort(label="a", hint="int", badge="nested"),),
+ outputs=(),
+ children=(),
+ edges=(),
+ )
+ source = render.render(node).source
+ self.assertIn("nested", source)
+ self.assertNotIn(">int<", source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestRetrospectiveNote(unittest.TestCase):
+ def test_no_recorded_edges_note_appears_in_cluster_label(self):
+ """The known WfMS limitation surfaces as an italic note on the cluster."""
+ data = wfms.run_recipe(_for_recipe(), xs=[1, 2, 3])
+ graph = retrospective.build(data, depth=0)
+ source = render.render(graph).source
+ self.assertIn("no recorded edges", source)
+
+
+@unittest.skipUnless(_has_graphviz, "graphviz not installed")
+class TestActuallyRenders(unittest.TestCase):
+ @unittest.skipIf(shutil.which("dot") is None, "Graphviz `dot` binary not installed")
+ def test_dot_accepts_the_source(self):
+ """The DOT we emit must survive a real Graphviz parse, not just look right."""
+ digraph = render.render(prospective.build(_for_recipe(), depth=0))
+ self.assertTrue(digraph.pipe(format="svg").startswith(b"