From bda182be962a9f54953a5793c89dc7059c745839 Mon Sep 17 00:00:00 2001 From: Andi Hellmund Date: Mon, 2 Mar 2026 14:48:49 +0100 Subject: [PATCH] visualization topology --- src/archml/views/topology.py | 527 +++++++++++++++++++++++++++ tests/views/test_topology.py | 680 +++++++++++++++++++++++++++++++++++ 2 files changed, 1207 insertions(+) create mode 100644 src/archml/views/topology.py create mode 100644 tests/views/test_topology.py diff --git a/src/archml/views/topology.py b/src/archml/views/topology.py new file mode 100644 index 0000000..a6545ec --- /dev/null +++ b/src/archml/views/topology.py @@ -0,0 +1,527 @@ +# Copyright 2026 ArchML Contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Abstract visualization topology model for ArchML diagrams. + +This module defines an intermediate representation of a diagram's topology +that is independent of both the architecture domain model and any rendering +backend. The same topology drives SVG export, interactive Dash views, and +any future renderer. + +The topology model captures: + +- **VizNode** — an opaque box (leaf component, system, external actor, or + interface terminal). +- **VizBoundary** — a labelled visual container grouping child nodes and/or + nested sub-boundaries. Corresponds to a component or system whose internal + structure is expanded at this zoom level. +- **VizPort** — a named interface connection point on a node or boundary. + Every ``requires``/``provides`` declaration in the architecture model + becomes a port. Ports are the endpoints of edges. +- **VizEdge** — a directed connection between two ports, derived from an + ArchML ``connect`` statement. +- **VizDiagram** — the assembled complete topology: root boundary, peripheral + nodes (terminals + externals), and all edges. + +Geometry (positions, sizes, edge waypoints) is deliberately absent. A +separate layout step computes geometry and attaches it to a layout model. +This decoupling lets a single topology feed both headless layout algorithms +and interactive, event-driven frontends. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Literal + +from archml.model.entities import Component, InterfaceRef, System + +# ############### +# Public Interface +# ############### + +NodeKind = Literal["component", "system", "external_component", "external_system", "terminal"] +"""Semantic classification of a :class:`VizNode`.""" + +BoundaryKind = Literal["component", "system"] +"""Semantic classification of a :class:`VizBoundary`.""" + + +@dataclass +class VizPort: + """A named interface connection point on a :class:`VizNode` or :class:`VizBoundary`. + + Ports are the typed endpoints of :class:`VizEdge` connections. Each + ``requires`` or ``provides`` declaration on an architecture entity produces + one port. + + Attributes: + id: Diagram-unique stable identifier used by :class:`VizEdge` and + renderers (e.g. ``"ECommerce__OrderService.req.PaymentRequest"``). + node_id: ID of the owning :class:`VizNode` or :class:`VizBoundary`. + interface_name: Base interface name without the version suffix. + interface_version: Version string (e.g. ``"v2"``), or ``None`` for + unversioned interfaces. + direction: ``"requires"`` for input ports; ``"provides"`` for output + ports. + description: Optional human-readable description of the interface. + """ + + id: str + node_id: str + interface_name: str + interface_version: str | None + direction: Literal["requires", "provides"] + description: str | None = None + + +@dataclass +class VizNode: + """A renderable box representing a leaf entity or interface terminal. + + A *leaf entity* is a component, system, or external actor whose internal + structure is not expanded in this diagram — it appears as an opaque box. + A *terminal* represents one of the focus entity's own ``requires`` or + ``provides`` interfaces, anchored at the diagram boundary. + + Attributes: + id: Diagram-unique stable identifier (e.g. ``"ECommerce__OrderService"`` + or ``"terminal.req.OrderRequest"``). + label: Short display name — typically the entity mnemonic. + title: Human-readable title if distinct from *label*; ``None`` + otherwise. + kind: Semantic classification that determines default visual styling. + entity_path: ``::``-delimited qualified path for navigation and + deep-linking (e.g. ``"ECommerce::OrderService"``). Empty string + for terminal nodes. + description: Tooltip / hover text; ``None`` if absent. + tags: Arbitrary labels inherited from the ArchML model, used for + filtering and conditional styling. + ports: Interface ports belonging to this node (requires and provides). + """ + + id: str + label: str + title: str | None + kind: NodeKind + entity_path: str + description: str | None = None + tags: list[str] = field(default_factory=list) + ports: list[VizPort] = field(default_factory=list) + + +@dataclass +class VizBoundary: + """A labelled visual container grouping child nodes and/or sub-boundaries. + + Boundaries represent container entities — components or systems whose + internal structure is rendered at this zoom level. They are drawn with a + visible bounding box and a title label. + + Attributes: + id: Diagram-unique stable identifier. + label: Short display name. + title: Human-readable title if distinct from *label*; ``None`` + otherwise. + kind: ``"component"`` or ``"system"``. + entity_path: Qualified path for navigation and deep-linking. + description: Tooltip / hover text; ``None`` if absent. + tags: Arbitrary labels from the ArchML model. + ports: The boundary entity's own interface ports (its ``requires`` and + ``provides`` declarations). + children: Direct visual children — a mix of leaf :class:`VizNode` + instances and nested :class:`VizBoundary` instances for recursively + expanded sub-entities. + """ + + id: str + label: str + title: str | None + kind: BoundaryKind + entity_path: str + description: str | None = None + tags: list[str] = field(default_factory=list) + ports: list[VizPort] = field(default_factory=list) + children: list[VizNode | VizBoundary] = field(default_factory=list) + + +@dataclass +class VizEdge: + """A directed connection between two ports in the diagram. + + Edges correspond to ArchML ``connect`` statements. The direction follows + the ArchML convention: the *source* is the requiring side (initiator), the + *target* is the providing side (responder), and the arrow represents the + request direction. + + Attributes: + id: Diagram-unique stable identifier derived from the port IDs + (e.g. ``"edge.A.req.IFace--B.prov.IFace"``). + source_port_id: ID of the source :class:`VizPort` (a ``requires`` + port). + target_port_id: ID of the target :class:`VizPort` (a ``provides`` + port). + label: Human-readable label shown on the edge + (``"InterfaceName"`` or ``"InterfaceName@vN"``). + interface_name: Base interface name without version suffix. + interface_version: Version string, or ``None``. + protocol: Optional transport protocol annotation (e.g. ``"gRPC"``). + is_async: Whether the connection is asynchronous. + description: Optional human-readable description of the connection. + """ + + id: str + source_port_id: str + target_port_id: str + label: str + interface_name: str + interface_version: str | None = None + protocol: str | None = None + is_async: bool = False + description: str | None = None + + +@dataclass +class VizDiagram: + """Complete topology description of a visualization diagram. + + A :class:`VizDiagram` is built from a single *focus entity* (a + :class:`~archml.model.entities.Component` or + :class:`~archml.model.entities.System`). + + - The focus entity becomes the :attr:`root` :class:`VizBoundary`. + - Its direct children are placed inside the root as :class:`VizNode` + instances (opaque at this zoom level). + - The focus entity's own ``requires``/``provides`` interfaces appear as + terminal :class:`VizNode` instances in :attr:`peripheral_nodes`. + - External actors that appear as connection endpoints but are not children + of the focus entity also appear in :attr:`peripheral_nodes`. + - All ArchML ``connect`` statements within the focus entity become + :class:`VizEdge` entries in :attr:`edges`. + + Geometry (positions, sizes, edge routes) is not part of this model; it is + computed by a separate layout step. + + Attributes: + id: Stable diagram identifier derived from the focus entity path + (e.g. ``"diagram.ECommerce"``). + title: Display title for the diagram. + description: Optional longer description of the focus entity. + root: The focus entity rendered as a :class:`VizBoundary`. + peripheral_nodes: Terminal and external :class:`VizNode` instances + that appear outside the root boundary. + edges: All directed connections visible in the diagram. + """ + + id: str + title: str + description: str | None + root: VizBoundary + peripheral_nodes: list[VizNode] = field(default_factory=list) + edges: list[VizEdge] = field(default_factory=list) + + +def build_viz_diagram( + entity: Component | System, + *, + external_entities: dict[str, Component | System] | None = None, +) -> VizDiagram: + """Build a :class:`VizDiagram` topology from a model entity. + + The *entity* becomes the root :class:`VizBoundary`. Its direct children + are placed inside the boundary as opaque :class:`VizNode` instances. + + The entity's own ``requires``/``provides`` interfaces become *terminal* + :class:`VizNode` instances in ``peripheral_nodes`` — one node per + interface, positioned at the diagram boundary. + + External actors that appear in ``connect`` statements but are not direct + children of *entity* are also appended to ``peripheral_nodes``. When + *external_entities* supplies model data for an actor, the resulting node + carries full metadata (title, description, tags, ports); otherwise a + minimal stub is created. + + If a ``connect`` statement references an interface that is not declared as + a port on either endpoint, an implicit port is created on that node so + that the edge can always be connected. + + Args: + entity: The focus component or system to visualize. + external_entities: Optional mapping from entity name to model entity + for resolving external connection endpoints. Only names that do + not match a direct child of *entity* are consulted. + + Returns: + A :class:`VizDiagram` describing the full diagram topology. + """ + ext = external_entities or {} + entity_path = entity.qualified_name or entity.name + root_id = _make_id(entity_path) + + # --- Child nodes (direct children rendered as opaque boxes) --- + child_node_map: dict[str, VizNode] = {} + for comp in entity.components: + child_path = f"{entity_path}::{comp.name}" + child_node_map[comp.name] = _make_child_node(comp, child_path) + + if isinstance(entity, System): + for sys in entity.systems: + child_path = f"{entity_path}::{sys.name}" + child_node_map[sys.name] = _make_child_node(sys, child_path) + + # --- Root boundary --- + root_ports = _make_ports(root_id, entity) + root = VizBoundary( + id=root_id, + label=entity.name, + title=entity.title, + kind="component" if isinstance(entity, Component) else "system", + entity_path=entity_path, + description=entity.description, + tags=list(entity.tags), + ports=root_ports, + children=list(child_node_map.values()), + ) + + # --- Peripheral nodes --- + peripheral_nodes: list[VizNode] = [] + + # Terminals: the focus entity's own interface boundary points. + for ref in entity.requires: + peripheral_nodes.append(_make_terminal_node(ref, "requires")) + for ref in entity.provides: + peripheral_nodes.append(_make_terminal_node(ref, "provides")) + + # External endpoints: actors referenced in connections but not children. + all_child_names = set(child_node_map) + external_node_map: dict[str, VizNode] = {} + for conn in entity.connections: + for ep_name in (conn.source.entity, conn.target.entity): + if ep_name in all_child_names or ep_name in external_node_map: + continue + ext_model = ext.get(ep_name) + ext_node = _make_external_node(ep_name, ext_model) + external_node_map[ep_name] = ext_node + peripheral_nodes.append(ext_node) + + # --- Edges (from explicit connect statements) --- + all_node_map: dict[str, VizNode] = {**child_node_map, **external_node_map} + edges: list[VizEdge] = [] + + for conn in entity.connections: + src_node = all_node_map.get(conn.source.entity) + tgt_node = all_node_map.get(conn.target.entity) + if src_node is None or tgt_node is None: + # Endpoint not resolvable — skip the edge rather than crashing. + continue + + src_port_id = _find_port_id(src_node, "requires", conn.interface) + tgt_port_id = _find_port_id(tgt_node, "provides", conn.interface) + + if src_port_id is None: + # Interface not explicitly declared — create an implicit port. + p = _make_port(src_node.id, "requires", conn.interface) + src_node.ports.append(p) + src_port_id = p.id + + if tgt_port_id is None: + p = _make_port(tgt_node.id, "provides", conn.interface) + tgt_node.ports.append(p) + tgt_port_id = p.id + + label = _iref_label(conn.interface) + edges.append( + VizEdge( + id=f"edge.{src_port_id}--{tgt_port_id}", + source_port_id=src_port_id, + target_port_id=tgt_port_id, + label=label, + interface_name=conn.interface.name, + interface_version=conn.interface.version, + protocol=conn.protocol, + is_async=conn.is_async, + description=conn.description, + ) + ) + + return VizDiagram( + id=f"diagram.{root_id}", + title=entity.title or entity.name, + description=entity.description, + root=root, + peripheral_nodes=peripheral_nodes, + edges=edges, + ) + + +def collect_all_ports(diagram: VizDiagram) -> dict[str, VizPort]: + """Return a flat ``port_id → VizPort`` mapping for the entire diagram. + + Traverses the root boundary (including any nested sub-boundaries), all + peripheral nodes, and the root boundary's own ports. + + Args: + diagram: The diagram to collect ports from. + + Returns: + Dictionary mapping each port's stable ID to the + :class:`VizPort` instance. + """ + result: dict[str, VizPort] = {} + _collect_boundary_ports(diagram.root, result) + for node in diagram.peripheral_nodes: + for p in node.ports: + result[p.id] = p + return result + + +# ################ +# Implementation +# ################ + + +def _make_id(entity_path: str) -> str: + """Convert a ``::``-delimited entity path to a DOM/URL-safe element ID.""" + return entity_path.replace("::", "__") + + +def _iref_label(ref: InterfaceRef) -> str: + """Return a display label for an interface reference.""" + return f"{ref.name}@{ref.version}" if ref.version else ref.name + + +def _port_id(node_id: str, direction: Literal["requires", "provides"], ref: InterfaceRef) -> str: + """Construct a stable port ID from its owner, direction, and interface.""" + dir_tag = "req" if direction == "requires" else "prov" + suffix = f"{ref.name}@{ref.version}" if ref.version else ref.name + return f"{node_id}.{dir_tag}.{suffix}" + + +def _make_port( + node_id: str, + direction: Literal["requires", "provides"], + ref: InterfaceRef, +) -> VizPort: + """Create a :class:`VizPort` for a single interface reference.""" + return VizPort( + id=_port_id(node_id, direction, ref), + node_id=node_id, + interface_name=ref.name, + interface_version=ref.version, + direction=direction, + ) + + +def _make_ports(node_id: str, entity: Component | System) -> list[VizPort]: + """Create :class:`VizPort` instances for all requires and provides of *entity*.""" + ports: list[VizPort] = [] + for ref in entity.requires: + ports.append(_make_port(node_id, "requires", ref)) + for ref in entity.provides: + ports.append(_make_port(node_id, "provides", ref)) + return ports + + +def _make_child_node(entity: Component | System, entity_path: str) -> VizNode: + """Create a :class:`VizNode` for a direct child of the focus entity.""" + node_id = _make_id(entity_path) + if isinstance(entity, Component): + kind: NodeKind = "external_component" if entity.is_external else "component" + else: + kind = "external_system" if entity.is_external else "system" + return VizNode( + id=node_id, + label=entity.name, + title=entity.title, + kind=kind, + entity_path=entity_path, + description=entity.description, + tags=list(entity.tags), + ports=_make_ports(node_id, entity), + ) + + +def _make_external_node(name: str, entity: Component | System | None) -> VizNode: + """Create a :class:`VizNode` for an external connection endpoint. + + When *entity* is provided its full metadata is used; otherwise a minimal + stub is created so the edge can still be represented. + """ + if entity is not None: + path = entity.qualified_name or name + node_id = _make_id(path) + kind: NodeKind = "external_component" if isinstance(entity, Component) else "external_system" + return VizNode( + id=node_id, + label=entity.name, + title=entity.title, + kind=kind, + entity_path=path, + description=entity.description, + tags=list(entity.tags), + ports=_make_ports(node_id, entity), + ) + # Stub for an endpoint whose model entity is unavailable. + node_id = f"ext.{name}" + return VizNode( + id=node_id, + label=name, + title=None, + kind="external_component", + entity_path=name, + ) + + +def _make_terminal_node( + ref: InterfaceRef, + direction: Literal["requires", "provides"], +) -> VizNode: + """Create a terminal :class:`VizNode` for the focus entity's own interface port. + + A terminal node anchors the focus entity's external interface boundary + visually: ``requires`` terminals appear on the input side of the diagram, + ``provides`` terminals on the output side. Each terminal carries exactly + one port mirroring the interface direction. + """ + label = _iref_label(ref) + dir_tag = "req" if direction == "requires" else "prov" + node_id = f"terminal.{dir_tag}.{label}" + port = VizPort( + id=f"{node_id}.port", + node_id=node_id, + interface_name=ref.name, + interface_version=ref.version, + direction=direction, + ) + return VizNode( + id=node_id, + label=label, + title=None, + kind="terminal", + entity_path="", + ports=[port], + ) + + +def _find_port_id( + node: VizNode, + direction: Literal["requires", "provides"], + ref: InterfaceRef, +) -> str | None: + """Return the port ID matching *direction* and *ref* on *node*, or ``None``.""" + for p in node.ports: + if p.direction == direction and p.interface_name == ref.name and p.interface_version == ref.version: + return p.id + return None + + +def _collect_boundary_ports(boundary: VizBoundary, result: dict[str, VizPort]) -> None: + """Recursively collect all ports from *boundary* and its children into *result*.""" + for p in boundary.ports: + result[p.id] = p + for child in boundary.children: + if isinstance(child, VizBoundary): + _collect_boundary_ports(child, result) + else: + for p in child.ports: + result[p.id] = p diff --git a/tests/views/test_topology.py b/tests/views/test_topology.py new file mode 100644 index 0000000..787dddd --- /dev/null +++ b/tests/views/test_topology.py @@ -0,0 +1,680 @@ +# Copyright 2026 ArchML Contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the abstract visualization topology model and its builder.""" + +from archml.model.entities import Component, Connection, ConnectionEndpoint, InterfaceRef, System +from archml.views.topology import ( + VizBoundary, + VizNode, + VizPort, + build_viz_diagram, + collect_all_ports, +) + +# ############### +# Helpers +# ############### + + +def _iref(name: str, version: str | None = None) -> InterfaceRef: + return InterfaceRef(name=name, version=version) + + +def _conn( + source: str, + target: str, + interface: str, + version: str | None = None, + protocol: str | None = None, + is_async: bool = False, + description: str | None = None, +) -> Connection: + return Connection( + source=ConnectionEndpoint(entity=source), + target=ConnectionEndpoint(entity=target), + interface=InterfaceRef(name=interface, version=version), + protocol=protocol, + is_async=is_async, + description=description, + ) + + +def _port_ids(ports: list[VizPort]) -> set[str]: + return {p.id for p in ports} + + +def _node_ids(nodes: list[VizNode]) -> set[str]: + return {n.id for n in nodes} + + +def _child_ids(boundary: VizBoundary) -> set[str]: + return {c.id for c in boundary.children} + + +# ############### +# VizDiagram — root boundary +# ############### + + +def test_root_boundary_id_from_entity_name() -> None: + """Root boundary ID is derived from the entity name.""" + comp = Component(name="Worker") + diag = build_viz_diagram(comp) + assert diag.root.id == "Worker" + + +def test_root_boundary_id_uses_qualified_name() -> None: + """When qualified_name is set it is preferred over plain name.""" + comp = Component(name="Worker", qualified_name="System::Worker") + diag = build_viz_diagram(comp) + assert diag.root.id == "System__Worker" + + +def test_root_boundary_coloncolon_replaced_by_double_underscore() -> None: + """``::`` separators in entity paths are replaced by ``__`` in IDs.""" + comp = Component(name="A", qualified_name="X::Y::A") + diag = build_viz_diagram(comp) + assert diag.root.id == "X__Y__A" + + +def test_root_boundary_label_and_title() -> None: + """Root boundary label is the entity mnemonic; title is the human name.""" + comp = Component(name="order_service", title="Order Service") + diag = build_viz_diagram(comp) + assert diag.root.label == "order_service" + assert diag.root.title == "Order Service" + + +def test_root_boundary_kind_component() -> None: + comp = Component(name="C") + diag = build_viz_diagram(comp) + assert diag.root.kind == "component" + + +def test_root_boundary_kind_system() -> None: + sys = System(name="S") + diag = build_viz_diagram(sys) + assert diag.root.kind == "system" + + +def test_root_boundary_description_propagated() -> None: + comp = Component(name="C", description="Does things") + diag = build_viz_diagram(comp) + assert diag.root.description == "Does things" + + +def test_root_boundary_tags_propagated() -> None: + comp = Component(name="C", tags=["critical", "pci"]) + diag = build_viz_diagram(comp) + assert diag.root.tags == ["critical", "pci"] + + +# ############### +# VizDiagram — metadata +# ############### + + +def test_diagram_id_prefixed() -> None: + comp = Component(name="Worker") + diag = build_viz_diagram(comp) + assert diag.id == "diagram.Worker" + + +def test_diagram_title_from_entity_title() -> None: + comp = Component(name="w", title="Worker") + diag = build_viz_diagram(comp) + assert diag.title == "Worker" + + +def test_diagram_title_falls_back_to_name() -> None: + comp = Component(name="Worker") + diag = build_viz_diagram(comp) + assert diag.title == "Worker" + + +def test_diagram_description_none_when_absent() -> None: + comp = Component(name="C") + diag = build_viz_diagram(comp) + assert diag.description is None + + +# ############### +# Root boundary — child nodes +# ############### + + +def test_child_component_becomes_viz_node() -> None: + """Each direct component child becomes a VizNode inside the root boundary.""" + child = Component(name="Alpha") + parent = Component(name="Parent", components=[child]) + diag = build_viz_diagram(parent) + ids = _child_ids(diag.root) + assert "Parent__Alpha" in ids + + +def test_child_system_becomes_viz_node() -> None: + """Each direct sub-system becomes a VizNode inside the root boundary.""" + sub = System(name="Sub") + parent = System(name="Root", systems=[sub]) + diag = build_viz_diagram(parent) + ids = _child_ids(diag.root) + assert "Root__Sub" in ids + + +def test_child_node_kind_component() -> None: + child = Component(name="C") + parent = System(name="S", components=[child]) + diag = build_viz_diagram(parent) + node = next(n for n in diag.root.children if isinstance(n, VizNode) and n.label == "C") + assert node.kind == "component" + + +def test_child_node_kind_system() -> None: + sub = System(name="Sub") + parent = System(name="Root", systems=[sub]) + diag = build_viz_diagram(parent) + node = next(n for n in diag.root.children if isinstance(n, VizNode) and n.label == "Sub") + assert node.kind == "system" + + +def test_external_child_node_kind() -> None: + """An external component child gets kind 'external_component'.""" + ext_comp = Component(name="ExtC", is_external=True) + parent = System(name="S", components=[ext_comp]) + diag = build_viz_diagram(parent) + node = next(n for n in diag.root.children if isinstance(n, VizNode) and n.label == "ExtC") + assert node.kind == "external_component" + + +def test_child_node_entity_path() -> None: + child = Component(name="Alpha") + parent = Component(name="Parent", components=[child]) + diag = build_viz_diagram(parent) + node = next(n for n in diag.root.children if isinstance(n, VizNode)) + assert node.entity_path == "Parent::Alpha" + + +def test_child_node_description_and_tags() -> None: + child = Component(name="C", description="desc", tags=["t1"]) + parent = System(name="S", components=[child]) + diag = build_viz_diagram(parent) + node = next(n for n in diag.root.children if isinstance(n, VizNode)) + assert node.description == "desc" + assert node.tags == ["t1"] + + +def test_leaf_entity_has_no_children() -> None: + comp = Component(name="Leaf") + diag = build_viz_diagram(comp) + assert diag.root.children == [] + + +# ############### +# Ports — root boundary +# ############### + + +def test_root_boundary_requires_port() -> None: + """Each ``requires`` declaration on the focus entity becomes a requires port.""" + comp = Component(name="C", requires=[_iref("DataFeed")]) + diag = build_viz_diagram(comp) + ports = {p.interface_name: p for p in diag.root.ports} + assert "DataFeed" in ports + assert ports["DataFeed"].direction == "requires" + + +def test_root_boundary_provides_port() -> None: + """Each ``provides`` declaration on the focus entity becomes a provides port.""" + comp = Component(name="C", provides=[_iref("Result")]) + diag = build_viz_diagram(comp) + ports = {p.interface_name: p for p in diag.root.ports} + assert "Result" in ports + assert ports["Result"].direction == "provides" + + +def test_root_port_versioned_interface() -> None: + comp = Component(name="C", provides=[_iref("API", version="v2")]) + diag = build_viz_diagram(comp) + port = diag.root.ports[0] + assert port.interface_version == "v2" + assert "v2" in port.id + + +def test_root_port_node_id_matches_root_id() -> None: + comp = Component(name="C", requires=[_iref("X")]) + diag = build_viz_diagram(comp) + assert diag.root.ports[0].node_id == diag.root.id + + +# ############### +# Ports — child nodes +# ############### + + +def test_child_node_ports_created_from_requires_provides() -> None: + """A child node carries ports for all its requires and provides.""" + child = Component(name="W", requires=[_iref("In")], provides=[_iref("Out")]) + parent = System(name="S", components=[child]) + diag = build_viz_diagram(parent) + node = next(n for n in diag.root.children if isinstance(n, VizNode)) + directions = {p.direction for p in node.ports} + assert "requires" in directions + assert "provides" in directions + + +def test_child_port_id_contains_node_id_and_interface() -> None: + child = Component(name="W", requires=[_iref("Feed")]) + parent = System(name="S", components=[child]) + diag = build_viz_diagram(parent) + node = next(n for n in diag.root.children if isinstance(n, VizNode)) + port = node.ports[0] + assert node.id in port.id + assert "Feed" in port.id + + +# ############### +# Peripheral nodes — terminals +# ############### + + +def test_requires_terminal_node_created() -> None: + """The focus entity's ``requires`` interfaces become terminal nodes.""" + comp = Component(name="C", requires=[_iref("OrderRequest")]) + diag = build_viz_diagram(comp) + ids = _node_ids(diag.peripheral_nodes) + assert "terminal.req.OrderRequest" in ids + + +def test_provides_terminal_node_created() -> None: + """The focus entity's ``provides`` interfaces become terminal nodes.""" + comp = Component(name="C", provides=[_iref("OrderConfirmation")]) + diag = build_viz_diagram(comp) + ids = _node_ids(diag.peripheral_nodes) + assert "terminal.prov.OrderConfirmation" in ids + + +def test_terminal_node_kind() -> None: + comp = Component(name="C", requires=[_iref("X")]) + diag = build_viz_diagram(comp) + terminal = diag.peripheral_nodes[0] + assert terminal.kind == "terminal" + + +def test_terminal_node_has_one_port() -> None: + comp = Component(name="C", requires=[_iref("X")]) + diag = build_viz_diagram(comp) + terminal = diag.peripheral_nodes[0] + assert len(terminal.ports) == 1 + + +def test_terminal_port_direction_matches_interface_direction() -> None: + """A requires terminal has a requires port; a provides terminal has a provides port.""" + comp = Component(name="C", requires=[_iref("In")], provides=[_iref("Out")]) + diag = build_viz_diagram(comp) + by_id = {n.id: n for n in diag.peripheral_nodes} + req_terminal = by_id["terminal.req.In"] + prov_terminal = by_id["terminal.prov.Out"] + assert req_terminal.ports[0].direction == "requires" + assert prov_terminal.ports[0].direction == "provides" + + +def test_versioned_terminal_label_includes_version() -> None: + comp = Component(name="C", provides=[_iref("API", version="v2")]) + diag = build_viz_diagram(comp) + ids = _node_ids(diag.peripheral_nodes) + assert "terminal.prov.API@v2" in ids + + +def test_no_terminals_for_leaf_without_interfaces() -> None: + comp = Component(name="Isolated") + diag = build_viz_diagram(comp) + # No terminals; no external nodes either. + assert diag.peripheral_nodes == [] + + +# ############### +# Peripheral nodes — external actors +# ############### + + +def test_external_actor_stub_created_for_unresolved_endpoint() -> None: + """An endpoint not among the children creates a stub external node.""" + child = Component(name="A", requires=[_iref("IFace")]) + conn = _conn("A", "StripeAPI", "IFace") + parent = System(name="S", components=[child], connections=[conn]) + diag = build_viz_diagram(parent) + ext_ids = _node_ids(diag.peripheral_nodes) + assert "ext.StripeAPI" in ext_ids + + +def test_external_actor_stub_has_external_kind() -> None: + child = Component(name="A", requires=[_iref("IFace")]) + conn = _conn("A", "Ext", "IFace") + parent = System(name="S", components=[child], connections=[conn]) + diag = build_viz_diagram(parent) + ext_node = next(n for n in diag.peripheral_nodes if n.id == "ext.Ext") + assert ext_node.kind in ("external_component", "external_system") + + +def test_external_actor_resolved_from_external_entities() -> None: + """When external_entities provides model data it is used for the node.""" + child = Component(name="A", requires=[_iref("Pay")]) + stripe = Component(name="StripeAPI", title="Stripe", provides=[_iref("Pay")], is_external=True) + conn = _conn("A", "StripeAPI", "Pay") + parent = System(name="S", components=[child], connections=[conn]) + + diag = build_viz_diagram(parent, external_entities={"StripeAPI": stripe}) + + ext_node = next(n for n in diag.peripheral_nodes if n.label == "StripeAPI") + assert ext_node.title == "Stripe" + assert ext_node.kind == "external_component" + # Resolved node carries full ports. + assert any(p.interface_name == "Pay" for p in ext_node.ports) + + +def test_external_actor_appears_only_once_even_in_multiple_connections() -> None: + """The same external actor referenced twice produces a single peripheral node.""" + a = Component(name="A", requires=[_iref("X"), _iref("Y")]) + conn1 = _conn("A", "Ext", "X") + conn2 = _conn("A", "Ext", "Y") + parent = System(name="S", components=[a], connections=[conn1, conn2]) + diag = build_viz_diagram(parent) + ext_nodes = [n for n in diag.peripheral_nodes if n.label == "Ext"] + assert len(ext_nodes) == 1 + + +def test_child_not_added_to_peripheral_nodes() -> None: + """A direct child of the focus entity never appears in peripheral_nodes.""" + child_a = Component(name="A", requires=[_iref("IFace")]) + child_b = Component(name="B", provides=[_iref("IFace")]) + conn = _conn("A", "B", "IFace") + parent = System(name="S", components=[child_a, child_b], connections=[conn]) + diag = build_viz_diagram(parent) + peripheral_labels = {n.label for n in diag.peripheral_nodes} + assert "A" not in peripheral_labels + assert "B" not in peripheral_labels + + +# ############### +# Edges +# ############### + + +def test_edge_created_for_connection() -> None: + """A VizEdge is created for each connect statement.""" + a = Component(name="A", requires=[_iref("IFace")]) + b = Component(name="B", provides=[_iref("IFace")]) + conn = _conn("A", "B", "IFace") + parent = System(name="S", components=[a, b], connections=[conn]) + diag = build_viz_diagram(parent) + assert len(diag.edges) == 1 + + +def test_edge_label_is_interface_name() -> None: + a = Component(name="A", requires=[_iref("PayReq")]) + b = Component(name="B", provides=[_iref("PayReq")]) + conn = _conn("A", "B", "PayReq") + parent = System(name="S", components=[a, b], connections=[conn]) + diag = build_viz_diagram(parent) + assert diag.edges[0].label == "PayReq" + + +def test_edge_label_includes_version() -> None: + a = Component(name="A", requires=[_iref("API", "v2")]) + b = Component(name="B", provides=[_iref("API", "v2")]) + conn = _conn("A", "B", "API", version="v2") + parent = System(name="S", components=[a, b], connections=[conn]) + diag = build_viz_diagram(parent) + assert diag.edges[0].label == "API@v2" + + +def test_edge_source_port_is_requires_port() -> None: + """Edge source_port_id references a requires port on the source node.""" + a = Component(name="A", requires=[_iref("IFace")]) + b = Component(name="B", provides=[_iref("IFace")]) + conn = _conn("A", "B", "IFace") + parent = System(name="S", components=[a, b], connections=[conn]) + diag = build_viz_diagram(parent) + all_ports = collect_all_ports(diag) + src_port = all_ports[diag.edges[0].source_port_id] + assert src_port.direction == "requires" + assert src_port.interface_name == "IFace" + + +def test_edge_target_port_is_provides_port() -> None: + """Edge target_port_id references a provides port on the target node.""" + a = Component(name="A", requires=[_iref("IFace")]) + b = Component(name="B", provides=[_iref("IFace")]) + conn = _conn("A", "B", "IFace") + parent = System(name="S", components=[a, b], connections=[conn]) + diag = build_viz_diagram(parent) + all_ports = collect_all_ports(diag) + tgt_port = all_ports[diag.edges[0].target_port_id] + assert tgt_port.direction == "provides" + assert tgt_port.interface_name == "IFace" + + +def test_edge_source_and_target_port_owners() -> None: + """Source port belongs to the source node; target port to the target node.""" + a = Component(name="A", requires=[_iref("IFace")]) + b = Component(name="B", provides=[_iref("IFace")]) + conn = _conn("A", "B", "IFace") + parent = System(name="S", components=[a, b], connections=[conn]) + diag = build_viz_diagram(parent) + all_ports = collect_all_ports(diag) + edge = diag.edges[0] + assert all_ports[edge.source_port_id].node_id == "S__A" + assert all_ports[edge.target_port_id].node_id == "S__B" + + +def test_edge_protocol_and_async_propagated() -> None: + a = Component(name="A", requires=[_iref("X")]) + b = Component(name="B", provides=[_iref("X")]) + conn = _conn("A", "B", "X", protocol="gRPC", is_async=True, description="async call") + parent = System(name="S", components=[a, b], connections=[conn]) + diag = build_viz_diagram(parent) + edge = diag.edges[0] + assert edge.protocol == "gRPC" + assert edge.is_async is True + assert edge.description == "async call" + + +def test_multiple_edges_created() -> None: + a = Component(name="A", requires=[_iref("X"), _iref("Y")]) + b = Component(name="B", provides=[_iref("X")]) + c = Component(name="C", provides=[_iref("Y")]) + conns = [_conn("A", "B", "X"), _conn("A", "C", "Y")] + parent = System(name="S", components=[a, b, c], connections=conns) + diag = build_viz_diagram(parent) + assert len(diag.edges) == 2 + + +def test_edge_to_unknown_source_creates_stub_and_edge() -> None: + """An unknown source endpoint becomes a stub external node and the edge is kept.""" + b = Component(name="B", provides=[_iref("IFace")]) + conn = _conn("Ghost", "B", "IFace") + parent = System(name="S", components=[b], connections=[conn]) + diag = build_viz_diagram(parent) + # Stub node is created for "Ghost". + peripheral_labels = {n.label for n in diag.peripheral_nodes} + assert "Ghost" in peripheral_labels + # Edge is still produced (connecting stub → B). + assert len(diag.edges) == 1 + + +def test_edge_to_unknown_target_creates_stub_and_edge() -> None: + """An unknown target endpoint becomes a stub external node and the edge is kept.""" + a = Component(name="A", requires=[_iref("IFace")]) + conn = _conn("A", "Ghost", "IFace") + parent = System(name="S", components=[a], connections=[conn]) + diag = build_viz_diagram(parent) + peripheral_labels = {n.label for n in diag.peripheral_nodes} + assert "Ghost" in peripheral_labels + assert len(diag.edges) == 1 + + +# ############### +# Implicit ports +# ############### + + +def test_implicit_port_created_when_requires_missing() -> None: + """When the source lacks an explicit requires declaration an implicit port is added.""" + # A has no requires declarations at all. + a = Component(name="A") + b = Component(name="B", provides=[_iref("IFace")]) + conn = _conn("A", "B", "IFace") + parent = System(name="S", components=[a, b], connections=[conn]) + diag = build_viz_diagram(parent) + + assert len(diag.edges) == 1 + all_ports = collect_all_ports(diag) + src_port = all_ports[diag.edges[0].source_port_id] + assert src_port.direction == "requires" + assert src_port.interface_name == "IFace" + + +def test_implicit_port_created_when_provides_missing() -> None: + a = Component(name="A", requires=[_iref("IFace")]) + b = Component(name="B") # no provides + conn = _conn("A", "B", "IFace") + parent = System(name="S", components=[a, b], connections=[conn]) + diag = build_viz_diagram(parent) + + assert len(diag.edges) == 1 + all_ports = collect_all_ports(diag) + tgt_port = all_ports[diag.edges[0].target_port_id] + assert tgt_port.direction == "provides" + assert tgt_port.interface_name == "IFace" + + +# ############### +# collect_all_ports +# ############### + + +def test_collect_all_ports_includes_root_ports() -> None: + comp = Component(name="C", requires=[_iref("X")], provides=[_iref("Y")]) + diag = build_viz_diagram(comp) + all_ports = collect_all_ports(diag) + names = {p.interface_name for p in all_ports.values()} + assert "X" in names + assert "Y" in names + + +def test_collect_all_ports_includes_child_ports() -> None: + child = Component(name="W", requires=[_iref("Feed")]) + parent = System(name="S", components=[child]) + diag = build_viz_diagram(parent) + all_ports = collect_all_ports(diag) + names = {p.interface_name for p in all_ports.values()} + assert "Feed" in names + + +def test_collect_all_ports_includes_terminal_ports() -> None: + comp = Component(name="C", provides=[_iref("Out")]) + diag = build_viz_diagram(comp) + all_ports = collect_all_ports(diag) + names = {p.interface_name for p in all_ports.values()} + assert "Out" in names + + +def test_collect_all_ports_includes_external_node_ports() -> None: + child = Component(name="A", requires=[_iref("Pay")]) + stripe = Component(name="Stripe", provides=[_iref("Pay")], is_external=True) + conn = _conn("A", "Stripe", "Pay") + parent = System(name="S", components=[child], connections=[conn]) + diag = build_viz_diagram(parent, external_entities={"Stripe": stripe}) + all_ports = collect_all_ports(diag) + prov_ports = [p for p in all_ports.values() if p.direction == "provides" and p.interface_name == "Pay"] + assert len(prov_ports) >= 1 + + +def test_collect_all_ports_returns_unique_ids() -> None: + """All returned port IDs are distinct.""" + a = Component(name="A", requires=[_iref("X")]) + b = Component(name="B", provides=[_iref("X")]) + conn = _conn("A", "B", "X") + parent = System(name="S", components=[a, b], connections=[conn]) + diag = build_viz_diagram(parent) + all_ports = collect_all_ports(diag) + assert len(all_ports) == len(set(all_ports)) + + +# ############### +# End-to-end: full e-commerce example +# ############### + + +def test_ecommerce_system_topology() -> None: + """Integration test building a topology for the canonical e-commerce example.""" + order_svc = Component( + name="OrderService", + title="Order Service", + requires=[_iref("PaymentRequest"), _iref("InventoryCheck")], + provides=[_iref("OrderConfirmation")], + ) + payment_gw = Component( + name="PaymentGateway", + title="Payment Gateway", + tags=["critical", "pci-scope"], + requires=[_iref("PaymentRequest")], + provides=[_iref("PaymentResult")], + ) + inventory = Component( + name="InventoryManager", + title="Inventory Manager", + requires=[_iref("InventoryCheck")], + provides=[_iref("InventoryStatus")], + ) + stripe = Component( + name="StripeAPI", + title="Stripe Payment API", + requires=[_iref("PaymentRequest")], + provides=[_iref("PaymentResult")], + is_external=True, + ) + + ecommerce = System( + name="ECommerce", + title="E-Commerce Platform", + components=[order_svc, payment_gw, inventory], + connections=[ + _conn("OrderService", "PaymentGateway", "PaymentRequest"), + _conn("OrderService", "InventoryManager", "InventoryCheck"), + _conn("PaymentGateway", "StripeAPI", "PaymentRequest", protocol="HTTP", is_async=True), + ], + ) + + diag = build_viz_diagram(ecommerce, external_entities={"StripeAPI": stripe}) + + # Root boundary is ECommerce. + assert diag.root.id == "ECommerce" + assert diag.root.kind == "system" + + # Three children inside the boundary. + child_labels = {c.label for c in diag.root.children if isinstance(c, VizNode)} + assert child_labels == {"OrderService", "PaymentGateway", "InventoryManager"} + + # StripeAPI is a peripheral (external) node. + peripheral_labels = {n.label for n in diag.peripheral_nodes} + assert "StripeAPI" in peripheral_labels + stripe_node = next(n for n in diag.peripheral_nodes if n.label == "StripeAPI") + assert stripe_node.kind == "external_component" + + # Three edges. + assert len(diag.edges) == 3 + edge_labels = {e.label for e in diag.edges} + assert "PaymentRequest" in edge_labels + assert "InventoryCheck" in edge_labels + + # Async annotation on the Stripe edge. + stripe_edge = next(e for e in diag.edges if e.protocol == "HTTP") + assert stripe_edge.is_async is True + + # All ports resolvable. + all_ports = collect_all_ports(diag) + for edge in diag.edges: + assert edge.source_port_id in all_ports + assert edge.target_port_id in all_ports