This repository has been archived on 2026-08-04. You can view files and clone it, but cannot push or open issues or pull requests.
orchestrator/agent-team/tests/test_topology.py
Adam Moussa e3137e33e6 test(agent-team): cover transitions, topology, dashboard API + rework status_page tests
Add test_transitions.py (recorder idempotency under replay, terminal close,
fail-soft, the B4 _instrument signature-preservation guarantee, end-to-end graph
drive), test_topology.py (meta coverage, tree grouping, edge classification,
phase->node map), and test_dashboard.py (/api/state contract + per-node state,
/api/topology, /api/task timeline+cost+partial+thread_id validation, all via
TestClient, fastapi-skipif guarded). Rework test_status_page.py to the data layer
(drop retired render_html/SVG tests). conftest exposes tests/ for sibling imports.
Frontend: Vitest+RTL for layout, TaskList filters/selection, TaskDrawer timeline.
2026-06-23 17:15:31 -04:00

60 lines
2.3 KiB
Python

"""Tests for the LangGraph-introspected pipeline topology (``agent_team.topology``)."""
from __future__ import annotations
from agent_team.topology import (
NODE_META,
PHASE_TO_NODE,
build_topology,
missing_meta,
node_for_phase,
)
def test_every_graph_node_has_meta() -> None:
"""A newly added graph node must fail loudly until its NODE_META is filled in."""
assert missing_meta() == []
def test_topology_has_expected_nodes_and_trees() -> None:
topo = build_topology()
ids = {n["id"] for n in topo["nodes"]}
# The maximal P3+ wiring nodes, derived from the real graph.
assert {"intake", "clarify", "plan", "review", "build_node", "verify_node"} <= ids
# __start__/__end__ are not drawn.
assert "__start__" not in ids and "__end__" not in ids
# intake roots the core tree; the SDLC pipeline branches off it.
tree_ids = {t["id"] for t in topo["trees"]}
assert {"core", "sdlc"} <= tree_ids
by_id = {n["id"]: n for n in topo["nodes"]}
assert by_id["intake"]["tree"] == "core"
assert by_id["clarify"]["kind"] == "gate" # human gate lives at clarify
# P3 nodes render but are flagged inert.
assert by_id["build_node"]["gated"] is True
def test_edges_classified_spine_branch_loopback() -> None:
topo = build_topology()
kinds = {(e["from"], e["to"]): e["kind"] for e in topo["edges"]}
assert kinds[("intake", "clarify")] == "spine"
assert kinds[("clarify", "plan")] == "spine"
# review -> plan is the revise loop; verify -> build_node is the build loop.
assert kinds[("review", "plan")] == "loopback"
assert kinds[("verify_node", "build_node")] == "loopback"
# No edge references the framework terminals.
nodes = {n["id"] for n in topo["nodes"]}
for e in topo["edges"]:
assert e["from"] in nodes and e["to"] in nodes
def test_phase_to_node_mapping() -> None:
assert node_for_phase("build") == "build_node"
assert node_for_phase("verify") == "verify_node"
assert node_for_phase("clarify") == "clarify"
# Terminal/exception phases map to no node.
assert node_for_phase("done") is None
assert node_for_phase("parked") is None
assert node_for_phase("") is None
# Every mapped target is a real meta node.
for node_id in PHASE_TO_NODE.values():
assert node_id in NODE_META