backend/tests/test_api.py

"""API surface: endpoints, validation, WebSocket streaming and the fallback path."""

from __future__ import annotations

import json

import pytest
from fastapi.testclient import TestClient

from flowtwin.main import app


@pytest.fixture(scope="module")
def client():
    with TestClient(app) as c:
        yield c


@pytest.fixture
def session(client):
    res = client.post("/api/simulation/start", json={
        "venue_id": "circuit_alpha",
        "scenario_id": "circuit_alpha_post_race",
        "crowd_size": 6000,
        "speed": 10,
    })
    assert res.status_code == 200, res.text
    sid = res.json()["session"]["session_id"]
    yield sid
    client.delete(f"/api/simulation/{sid}")


# ── metadata ─────────────────────────────────────────────────────────

def test_healthz(client):
    body = client.get("/healthz").json()
    assert body["status"] == "ok"


def test_meta_reports_prediction_and_perception(client):
    body = client.get("/api/meta").json()
    assert body["name"] == "FlowTwin"
    assert "prediction" in body and "source" in body["prediction"]
    assert "perception" in body
    assert body["config"]["optimizer"]


def test_venue_and_scenario_listing(client):
    venues = client.get("/api/venues").json()["venues"]
    ids = {v["id"] for v in venues}
    assert {"circuit_alpha", "barcelona_2022"} <= ids

    scenarios = client.get("/api/scenarios").json()["scenarios"]
    assert scenarios[0]["id"] == "circuit_alpha_post_race", \
        "Simulation 1 must be presented first"
    assert any(s["id"] == "barcelona_2022_egress" for s in scenarios)


def test_barcelona_carries_its_provenance(client):
    venue = client.get("/api/venues/barcelona_2022").json()
    prov = venue["provenance"]
    assert prov["facts"] and prov["assumptions"]
    assert "counterfactual" in prov["disclaimer"].lower()
    for fact in prov["facts"]:
        assert fact["source"], "a documented fact must cite a source"


def test_unknown_venue_is_404(client):
    assert client.get("/api/venues/atlantis").status_code == 404


# ── simulation lifecycle ─────────────────────────────────────────────

def test_start_returns_a_usable_first_frame(client, session):
    frame = client.get(f"/api/simulation/{session}/state").json()
    assert frame["type"] == "frame"
    assert frame["t_s"] == 0.0
    assert len(frame["edges"]) > 0
    assert len(frame["nodes"]) > 0
    assert frame["metrics"]["agents_total"] == 6000
    assert frame["prediction"]["source"] in {"trained_model", "analytic_baseline"}


def test_scenario_venue_mismatch_is_rejected(client):
    res = client.post("/api/simulation/start", json={
        "venue_id": "circuit_alpha",
        "scenario_id": "barcelona_2022_egress",
    })
    assert res.status_code == 400


def test_invalid_inputs_fail_gracefully(client):
    bad = [
        {"venue_id": "circuit_alpha", "scenario_id": "circuit_alpha_post_race", "speed": 7},
        {"venue_id": "circuit_alpha", "scenario_id": "circuit_alpha_post_race", "crowd_size": 5},
        {"venue_id": "circuit_alpha", "scenario_id": "circuit_alpha_post_race",
         "routing_policy": "telepathy"},
        {"venue_id": "circuit_alpha", "scenario_id": "circuit_alpha_post_race",
         "capacity_overrides": {"EXIT_B": 99.0}},
        {"venue_id": "circuit_alpha", "scenario_id": "nope"},
    ]
    for payload in bad:
        res = client.post("/api/simulation/start", json=payload)
        assert res.status_code in (400, 404, 422), f"{payload} -> {res.status_code}"


def test_event_factor_override_is_honoured(client):
    res = client.post("/api/simulation/start", json={
        "venue_id": "circuit_alpha",
        "scenario_id": "circuit_alpha_post_race",
        "crowd_size": 3000,
        "event_factor_overrides": {"EXIT_B": 0.25},
    })
    assert res.status_code == 200, res.text
    sid = res.json()["session"]["session_id"]
    try:
        client.post(f"/api/simulation/{sid}/control",
                    json={"action": "run_to", "target_time_s": 300})
        frame = client.get(f"/api/simulation/{sid}/state?agents=false").json()
        exit_b = next(n for n in frame["nodes"] if n["id"] == "EXIT_B")
        assert exit_b["cap_pct"] == 25
    finally:
        client.delete(f"/api/simulation/{sid}")


def test_unknown_session_is_404(client):
    assert client.get("/api/simulation/deadbeef/state").status_code == 404
    assert client.post("/api/simulation/deadbeef/control",
                       json={"action": "play"}).status_code == 404


def test_control_actions(client, session):
    assert client.post(f"/api/simulation/{session}/control",
                       json={"action": "play"}).json()["session"]["playing"] is True
    assert client.post(f"/api/simulation/{session}/control",
                       json={"action": "pause"}).json()["session"]["playing"] is False
    assert client.post(f"/api/simulation/{session}/control",
                       json={"action": "speed", "speed": 20}).json()["session"]["speed"] == 20

    before = client.get(f"/api/simulation/{session}/state?agents=false").json()["t_s"]
    client.post(f"/api/simulation/{session}/control", json={"action": "step", "seconds": 60})
    after = client.get(f"/api/simulation/{session}/state?agents=false").json()["t_s"]
    assert after > before

    client.post(f"/api/simulation/{session}/control",
                json={"action": "run_to", "target_time_s": 400})
    assert client.get(f"/api/simulation/{session}/state?agents=false").json()["t_s"] >= 400


def test_control_requires_its_arguments(client, session):
    assert client.post(f"/api/simulation/{session}/control",
                       json={"action": "speed"}).status_code == 400
    assert client.post(f"/api/simulation/{session}/control",
                       json={"action": "invent"}).status_code == 422


def test_scripted_event_fires_and_is_reported(client, session):
    client.post(f"/api/simulation/{session}/control",
                json={"action": "run_to", "target_time_s": 300})
    frame = client.get(f"/api/simulation/{session}/state?agents=false").json()
    labels = [e["label"] for e in frame["events"]]
    assert any("Exit B" in l for l in labels), labels
    exit_b = next(n for n in frame["nodes"] if n["id"] == "EXIT_B")
    assert exit_b["cap_pct"] == 50


# ── intelligence endpoints ───────────────────────────────────────────

def test_alerts_and_prediction_endpoints(client, session):
    client.post(f"/api/simulation/{session}/control",
                json={"action": "run_to", "target_time_s": 900})
    alerts = client.get(f"/api/simulation/{session}/alerts").json()
    assert "alerts" in alerts and "bottlenecks" in alerts
    pred = client.get(f"/api/simulation/{session}/prediction").json()
    assert pred["horizons"] and pred["top"]


def test_strategy_simulate_then_apply(client, session):
    client.post(f"/api/simulation/{session}/control",
                json={"action": "run_to", "target_time_s": 900})

    result = client.post(f"/api/simulation/{session}/strategy/simulate",
                         json={"horizon_s": 120}).json()
    assert result["available"], result
    assert len(result["strategies"]) >= 4
    assert result["bottleneck"]["base_id"]
    assert result["recommendation"]["reasons"]
    assert sum(1 for s in result["strategies"] if s["recommended"]) == 1

    # Scores must be ordered and the winner must be first.
    scores = [s["score"] for s in result["strategies"]]
    assert scores == sorted(scores)

    winner = result["recommendation"]["strategy_id"]
    applied = client.post(f"/api/simulation/{session}/strategy/apply",
                          json={"strategy_id": winner})
    assert applied.status_code == 200, applied.text
    assert applied.json()["agents_affected"] >= 0

    frame = client.get(f"/api/simulation/{session}/state?agents=false").json()
    assert frame["interventions"], "the applied intervention was not recorded"


def test_applying_an_unknown_strategy_is_rejected(client, session):
    client.post(f"/api/simulation/{session}/control",
                json={"action": "run_to", "target_time_s": 900})
    res = client.post(f"/api/simulation/{session}/strategy/apply",
                      json={"strategy_id": "nonsense"})
    assert res.status_code == 400


def test_optimize_is_an_alias_of_simulate(client, session):
    client.post(f"/api/simulation/{session}/control",
                json={"action": "run_to", "target_time_s": 900})
    res = client.post(f"/api/simulation/{session}/strategy/optimize", json={"horizon_s": 120})
    assert res.status_code == 200
    assert res.json()["available"]


# ── streaming ────────────────────────────────────────────────────────

def test_websocket_streams_frames(client, session):
    client.post(f"/api/simulation/{session}/control", json={"action": "play"})
    with client.websocket_connect(f"/api/ws/simulation/{session}") as ws:
        first = ws.receive_json()
        assert first["type"] == "frame"
        assert first["session_id"] == session
        seen = 0
        for _ in range(6):
            msg = ws.receive_json()
            if msg["type"] == "frame":
                seen += 1
                break
        assert seen >= 1, "no further frames were pushed"


def test_websocket_rejects_an_unknown_session(client):
    with client.websocket_connect("/api/ws/simulation/deadbeef") as ws:
        assert ws.receive_json()["type"] == "error"


# ── perception & benchmarks ──────────────────────────────────────────

def test_perception_status_is_always_answerable(client):
    body = client.get("/api/perception/status").json()
    assert "loaded" in body and "candidates" in body
    assert len(body["candidates"]) >= 2


def test_perception_never_invents_a_count(client):
    """With no model available the endpoint must fail loudly, not guess."""
    res = client.post("/api/perception/analyze", files={
        "file": ("x.png", b"not-an-image", "image/png")})
    assert res.status_code in (200, 503)
    if res.status_code == 200:
        assert res.json()["observation"]["people"] >= 0
    else:
        assert "detail" in res.json()


# ── demo fallback ────────────────────────────────────────────────────

def test_recorded_run_replays_through_the_same_interface(client):
    """The fallback must be indistinguishable from a live run to the dashboard."""
    res = client.post("/api/simulation/start", json={
        "venue_id": "circuit_alpha",
        "scenario_id": "circuit_alpha_post_race",
        "use_recording": True,
    })
    if res.status_code == 404:
        pytest.skip("no recording present; run scripts/record_fallback.py")
    assert res.status_code == 200, res.text
    sid = res.json()["session"]["session_id"]
    assert res.json()["session"]["kind"] == "replay"
    try:
        first = client.get(f"/api/simulation/{sid}/state").json()
        # Same frame shape as a live session — the frontend cannot tell.
        for key in ("edges", "nodes", "metrics", "alerts", "prediction",
                    "bottlenecks", "events"):
            assert key in first, f"replay frame is missing {key}"

        client.post(f"/api/simulation/{sid}/control",
                    json={"action": "run_to", "target_time_s": 900})
        later = client.get(f"/api/simulation/{sid}/state?agents=false").json()
        assert later["t_s"] >= 900
        assert any(a["severity"] == "critical" for a in later["alerts"])

        strategies = client.post(f"/api/simulation/{sid}/strategy/simulate",
                                 json={}).json()
        assert strategies["available"]
        assert len(strategies["strategies"]) >= 4
        assert strategies["recommendation"]["strategy_id"]

        applied = client.post(f"/api/simulation/{sid}/strategy/apply",
                              json={"strategy_id": strategies["recommendation"]["strategy_id"]})
        assert applied.status_code == 200
    finally:
        client.delete(f"/api/simulation/{sid}")


def test_benchmarks_endpoint(client):
    body = client.get("/api/benchmarks").json()
    assert "available" in body
    if body["available"]:
        scenarios = body["scenarios"]
        assert scenarios
        for payload in scenarios.values():
            assert payload["seeds"]
            assert payload["stats"]