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

from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)


def test_health_declares_real_persisted_mode() -> None:
    response = client.get("/api/health")
    assert response.status_code == 200
    assert response.json()["data_mode"] == "persisted-real-run"


def test_real_e2e_runs_900029_and_900031_are_discoverable() -> None:
    response = client.get("/api/runs")
    assert response.status_code == 200
    by_id = {run["script_build_id"]: run for run in response.json()}
    assert {900029, 900031} <= set(by_id)
    assert by_id[900029]["task_count"] >= 11
    assert by_id[900031]["task_count"] >= 8
    assert by_id[900029]["planner_turn_count"] > 10
    assert by_id[900031]["planner_turn_count"] > 10


def test_execution_projects_global_planner_and_dynamic_task_tree() -> None:
    for build_id in (900029, 900031):
        response = client.get(f"/api/runs/{build_id}/execution")
        assert response.status_code == 200
        body = response.json()
        assert body["data_mode"] == "persisted-real-run"
        assert len([task for task in body["tasks"] if task["parent_task_id"] is None]) == 1
        assert any(task["task_kind"] == "direction" for task in body["tasks"])
        assert any(task["task_kind"] == "compose" for task in body["tasks"])
        assert {turn["tool_name"] for turn in body["planner_turns"]} >= {
            "inspect_script_plan",
            "plan_script_tasks",
            "dispatch_script_tasks",
            "decide_script_task",
        }
        visible_counts = [len(turn["visible_task_ids"]) for turn in body["planner_turns"]]
        assert visible_counts == sorted(visible_counts)
        assert visible_counts[-1] == len(body["tasks"])


def test_task_detail_exposes_lifecycle_worker_loop_and_data_proof() -> None:
    execution = client.get("/api/runs/900029/execution").json()
    paragraph = next(task for task in execution["tasks"] if task["task_kind"] == "paragraph")
    response = client.get(f"/api/runs/900029/tasks/{paragraph['task_id']}")
    assert response.status_code == 200
    body = response.json()
    assert {event["kind"] for event in body["lifecycle"]} >= {"task", "attempt", "validation", "decision"}
    assert body["attempts"]
    steps = body["attempts"][0]["steps"]
    assert any(step["step_kind"] == "tool_call" for step in steps)
    assert body["attempts"][0]["prompt"].startswith("Produce one independently testable Paragraph")
    assert body["data_usage"]["confirmed_reads"]
    assert body["data_usage"]["produced_outputs"]


def test_unknown_run_and_task_are_404() -> None:
    assert client.get("/api/runs/999999/execution").status_code == 404
    assert client.get("/api/runs/900029/tasks/missing").status_code == 404