"""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"]