script-build-and-framework-tmp/visualization/backend/tests/test_api.py

from fastapi.testclient import TestClient

from app.contracts import SCHEMA_CATALOG
from app.fake_data import (
    ARTIFACT_DIGESTS,
    GRAPH,
    RUN_ID,
    canonical_digest,
    canonical_wire_digest,
)
from app.main import app

client = TestClient(app)


def test_health_declares_fake_mode() -> None:
    response = client.get("/api/health")
    assert response.status_code == 200
    assert response.json()["data_mode"] == "fake"


def test_graph_is_fine_grained_and_connected() -> None:
    response = client.get(f"/api/runs/{RUN_ID}/graph")
    assert response.status_code == 200
    body = response.json()
    assert body["data_mode"] == "fake"
    assert len(body["nodes"]) >= 55
    assert len(body["edges"]) >= 50
    assert {item["phase"] for item in body["nodes"]} == {
        "phase-1",
        "phase-2",
        "phase-3",
        "delivery",
    }
    assert {item["node_type"] for item in body["nodes"]} >= {
        "contract",
        "operation",
        "attempt",
        "artifact",
        "validation",
        "decision",
        "publication",
        "transaction",
        "readback",
    }


def test_every_fake_record_uses_exact_contract_fields() -> None:
    used_models: set[str] = set()
    for node in GRAPH.nodes:
        model = node.record.model_name
        used_models.add(model)
        assert set(node.record.payload) == set(SCHEMA_CATALOG[model]), node.id
    required = {
        "ScriptBuildRecord",
        "ScriptBuildInputSnapshotV1",
        "MissionBinding",
        "MissionOwnerToken",
        "ScriptTaskContractV1",
        "OperationView",
        "AttemptView",
        "ValidationView",
        "PlannerDecisionView",
        "EvidenceRecordV1",
        "ScriptDirectionArtifactV1",
        "StructureArtifactV1",
        "ParagraphArtifactV1",
        "ElementSetArtifactV1",
        "ComparisonArtifactV1",
        "StructuredScriptArtifactV1",
        "CandidatePortfolioArtifactV1",
        "RootDeliveryManifestV1",
        "LegacyProjectionCanonicalV1",
        "Publication",
        "PublicationResult",
        "HttpCommandRecord",
        "InputSnapshotRow",
        "ArtifactVersionRow",
        "ParagraphRow",
        "ElementRow",
        "ParagraphElementLinkRow",
    }
    assert required <= used_models


def test_manifest_and_legacy_projection_use_real_logical_closure() -> None:
    records = {node.record.model_name: node.record.payload for node in GRAPH.nodes}
    canonical = records["LegacyProjectionCanonicalV1"]
    manifest = records["RootDeliveryManifestV1"]
    assert manifest["legacy_projection_digest"] == canonical_wire_digest(canonical)
    assert manifest["input_closure_digest"] != manifest["legacy_projection_digest"]
    for paragraph in canonical["paragraphs"]:
        assert "logical_key" in paragraph
        assert "parent_logical_key" in paragraph
        assert not {"paragraph_id", "parent_id", "is_active"} & set(paragraph)
    for element in canonical["elements"]:
        assert "logical_key" in element
        assert not {"element_id", "is_active"} & set(element)
    for link in canonical["paragraph_element_links"]:
        assert set(link) == {"paragraph_logical_key", "element_logical_key"}


def test_artifact_refs_pin_specific_kind_and_canonical_digest() -> None:
    attempts = [
        node.record.payload
        for node in GRAPH.nodes
        if node.record.model_name == "AttemptView"
    ]
    for attempt in attempts:
        for ref in attempt["submission"]["artifact_refs"]:
            version = int(ref["version"])
            assert ref["kind"] != "business-artifact"
            assert ref["digest"] == ARTIFACT_DIGESTS[version]
    node_ids = {node.id for node in GRAPH.nodes}
    assert "compose:alternate:artifact" in node_ids
    assert "compose:alternate:decision" in node_ids


def test_root_task_has_no_self_parent_and_declares_children() -> None:
    root = next(
        node.record.payload
        for node in GRAPH.nodes
        if node.record.model_name == "TaskView"
        and node.record.payload["task_id"] == "task-root"
    )
    assert root["parent_task_id"] is None
    assert root["child_task_ids"] == ["direction", "portfolio"]
    contract = next(
        node.record.payload
        for node in GRAPH.nodes
        if node.id == "root-delivery:contract"
    )
    assert contract["scope_ref"] == "script-build://scope/root"
    assert contract["write_scope"] == []
    assert [item["expected_task_kind"] for item in contract["input_decision_refs"]] == [
        "direction",
        "candidate-portfolio",
    ]
    attempt = next(
        node.record.payload
        for node in GRAPH.nodes
        if node.id == "root-delivery:attempt"
    )
    assert attempt["accepted_child_decision_ids"] == [
        "decision-direction",
        "decision-portfolio",
    ]


def test_input_snapshot_row_and_logical_view_share_one_canonical_payload() -> None:
    row = next(
        node.record.payload
        for node in GRAPH.nodes
        if node.record.model_name == "InputSnapshotRow"
    )
    logical = next(
        node.record.payload
        for node in GRAPH.nodes
        if node.record.model_name == "ScriptBuildInputSnapshotV1"
    )
    assert logical["snapshot_id"] == str(row["id"]) == "301"
    assert row["canonical_sha256"] == canonical_digest(row["canonical_json"])
    assert logical["canonical_sha256"] == f"sha256:{row['canonical_sha256']}"
    for key in (
        "script_build_id",
        "execution_id",
        "topic_build_id",
        "topic_id",
        "topic",
        "account",
        "persona_points",
        "section_patterns",
        "strategies",
        "prompt_manifest",
        "datasource_manifest",
        "model_manifest",
    ):
        assert logical[key] == row["canonical_json"][key]


def test_final_physical_row_counts_match_canonical_counts() -> None:
    canonical = next(
        node.record.payload
        for node in GRAPH.nodes
        if node.record.model_name == "LegacyProjectionCanonicalV1"
    )
    counts = {
        model: sum(1 for node in GRAPH.nodes if node.record.model_name == model)
        for model in ("ParagraphRow", "ElementRow", "ParagraphElementLinkRow")
    }
    assert counts == {
        "ParagraphRow": canonical["paragraph_count"],
        "ElementRow": canonical["element_count"],
        "ParagraphElementLinkRow": canonical["link_count"],
    }


def test_node_detail_and_not_found() -> None:
    node_id = GRAPH.nodes[10].id
    response = client.get(f"/api/runs/{RUN_ID}/nodes/{node_id}")
    assert response.status_code == 200
    assert response.json()["node"]["id"] == node_id
    assert client.get(f"/api/runs/999/nodes/{node_id}").status_code == 404
    assert client.get(f"/api/runs/{RUN_ID}/nodes/missing").status_code == 404