railiance-fabric/tests/test_coordination_graph.py

from __future__ import annotations

from railiance_fabric.coordination_graph import coordination_graph_payload
from railiance_fabric.graph_explorer import fabric_graph_explorer_manifest
from railiance_fabric.schema_validation import draft202012_validator


def _validate_schema(name: str, document: dict) -> None:
    from pathlib import Path

    schema_path = Path("schemas") / name
    validator = draft202012_validator(schema_path)
    validator.validate(document)


def test_coordination_payload_filters_open_records_and_cites_wait() -> None:
    workplans = [
        {"id": "wp-a", "slug": "cust-wp-0071", "title": "Sizing", "status": "active"},
        {"id": "wp-b", "slug": "rapps-wp-0014", "title": "Pilot", "status": "active"},
        {"id": "wp-c", "slug": "done-wp", "title": "Finished", "status": "finished"},
    ]
    tasks = [
        {
            "id": "t1",
            "workplan_id": "wp-a",
            "title": "Measure after RAPPS-WP-0014",
            "status": "wait",
            "description": "Await RAPPS-WP-0014 bind",
            "needs_human": False,
        },
        {
            "id": "t2",
            "workplan_id": "wp-a",
            "title": "Closed",
            "status": "done",
        },
        {
            "id": "t3",
            "workplan_id": "wp-c",
            "title": "orphan open on finished wp",
            "status": "todo",
        },
    ]
    payload = coordination_graph_payload(workplans, tasks)
    _validate_schema("graph-explorer-payload.schema.yaml", payload)
    ids = {el["data"]["id"] for el in payload["elements"]}
    assert "workplan:wp-a" in ids
    assert "workplan:wp-b" in ids
    assert "workplan:wp-c" not in ids
    assert "task:t1" in ids
    assert "task:t2" not in ids
    assert "task:t3" not in ids
    waits = [el for el in payload["elements"] if el["data"].get("edgeType") == "waits_on"]
    assert waits
    assert waits[0]["data"]["target"] == "workplan:wp-b"
    assert payload["metrics"]["open_workplans"] == 2
    assert "coordination" in {mode["id"] for mode in fabric_graph_explorer_manifest()["modes"]}
    assert payload["metrics"]["wait_or_human_tasks"] == 1
    assert waits[0]["data"]["edgeSource"] == "citation"
    assert payload["metrics"]["depends_on_edges"] == 0
    assert payload["metrics"]["residual_open_workplans"] == 0


def test_residuals_omitted_unless_included_and_titles_are_not_consulted() -> None:
    workplans = [
        {
            "id": "wp-rel",
            "slug": "cust-wp-0072",
            "title": "Backfill",
            "status": "active",
            "flavor": "planning",
        },
        {
            "id": "wp-res",
            "slug": "fin-wp-0006",
            "title": "Consume settlement",
            "status": "proposed",
            "flavor": "residual",
        },
        {
            "id": "wp-named",
            "slug": "upc-wp-0002",
            "title": "Company vessel close-out (G1 + residual G2/G8)",
            "status": "active",
            "flavor": "implementation",
        },
    ]
    tasks = [
        {
            "id": "t-res",
            "workplan_id": "wp-res",
            "title": "Leftover task",
            "status": "todo",
            "flavor": "residual",
        }
    ]
    hidden = coordination_graph_payload(workplans, tasks)
    ids = {el["data"]["id"] for el in hidden["elements"]}
    assert "workplan:wp-rel" in ids
    assert "workplan:wp-named" in ids
    assert "workplan:wp-res" not in ids
    assert "task:t-res" not in ids
    assert hidden["metrics"]["residual_open_workplans"] == 1
    assert hidden["metrics"]["include_residuals"] is False

    shown = coordination_graph_payload(workplans, tasks, include_residuals=True)
    shown_ids = {el["data"]["id"] for el in shown["elements"]}
    assert "workplan:wp-res" in shown_ids
    assert "task:t-res" in shown_ids
    residual_node = next(
        el for el in shown["elements"] if el["data"]["id"] == "workplan:wp-res"
    )
    assert residual_node["data"]["flavor"] == "residual"
    assert residual_node["data"]["nodeClass"] == "residual"


def test_indexed_depends_on_outranks_citation_fallback() -> None:
    workplans = [
        {
            "id": "wp-a",
            "slug": "cust-wp-0072",
            "title": "Backfill",
            "status": "active",
            "flavor": "planning",
            "depends_on": [{"workplan_id": "wp-b", "workplan_slug": "state-wp-0092"}],
        },
        {
            "id": "wp-b",
            "slug": "state-wp-0092",
            "title": "Flavor schema",
            "status": "active",
            "flavor": "planning",
        },
    ]
    tasks = [
        {
            "id": "t1",
            "workplan_id": "wp-a",
            "title": "Wait for STATE-WP-0092",
            "status": "wait",
            "description": "Blocked on STATE-WP-0092",
        }
    ]
    payload = coordination_graph_payload(workplans, tasks)
    edges = [el["data"] for el in payload["elements"] if el["data"].get("kind") == "Edge"]
    depends = [e for e in edges if e.get("edgeType") == "depends_on"]
    waits = [e for e in edges if e.get("edgeType") == "waits_on"]
    assert len(depends) == 1
    assert depends[0]["source"] == "workplan:wp-a"
    assert depends[0]["target"] == "workplan:wp-b"
    assert depends[0]["edgeSource"] == "indexed"
    assert waits == []
    assert payload["metrics"]["depends_on_edges"] == 1
    assert payload["metrics"]["citation_edges"] == 0
    chokepoint = next(
        el["data"]["chokepoint"]
        for el in payload["elements"]
        if el["data"]["id"] == "workplan:wp-b"
    )
    assert chokepoint == 1


def test_canonical_id_depends_on_resolves_to_hub_uuid() -> None:
    workplans = [
        {
            "id": "uuid-a",
            "slug": "rail-fab-wp-0030",
            "title": "Graph",
            "status": "proposed",
            "depends_on": ["STATE-WP-0092"],
        },
        {
            "id": "uuid-b",
            "slug": "state-wp-0092",
            "title": "Schema",
            "status": "active",
        },
    ]
    payload = coordination_graph_payload(workplans, [])
    depends = [
        el["data"]
        for el in payload["elements"]
        if el["data"].get("edgeType") == "depends_on"
    ]
    assert depends
    assert depends[0]["target"] == "workplan:uuid-b"