festifun/backend/tests/test_api.py

"""API contract tests (spec M3), against a synthetic fixture built into the test data dir.

Locks the frozen response shapes lane C depends on, and verifies HTTP Range serving (206),
CORS, and graceful degradation of the still-stubbed endpoints.
"""

from __future__ import annotations

import shutil

import pytest

pytestmark = pytest.mark.skipif(
    shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None,
    reason="ffmpeg/ffprobe required to build the API test fixture",
)


@pytest.fixture(scope="module")
def client():
    from fastapi.testclient import TestClient

    from festival4d import synthetic

    synthetic.build(duration_s=1.0, run_ffmpeg=True)  # into the temp data dir (conftest)

    from festival4d import api  # imported after build so the /media mount + DB are ready

    with TestClient(api.app) as c:
        yield c


def test_manifest_shape(client):
    data = client.get("/api/manifest").json()
    # Exact-set lock: additive manifest fields are fine, but must be added here consciously
    # (this assertion has already caught two silent drifts — has_splat and has_capture).
    assert set(data) == {"videos", "t_global_max", "has_poses", "has_splat", "has_capture"}
    assert data["has_poses"] is True
    assert data["has_capture"] is False  # capture routes are opt-in (FESTIVAL4D_CAPTURE)
    assert len(data["videos"]) == 3
    v = data["videos"][0]
    assert set(v) == {
        "id", "filename", "url", "duration_s", "fps", "width", "height",
        "offset_ms", "drift_ppm",
    }
    offsets = {vv["filename"]: vv["offset_ms"] for vv in data["videos"]}
    assert offsets == {"cam0.mp4": 0.0, "cam1.mp4": 1370.0, "cam2.mp4": -842.0}


def test_poses_shape(client):
    poses = client.get("/api/videos/1/poses").json()
    assert len(poses) > 0
    p = poses[0]
    assert set(p) == {"frame_idx", "t_video_s", "q", "t", "intrinsics", "registered"}
    assert len(p["q"]) == 4 and len(p["t"]) == 3
    assert set(p["intrinsics"]) == {"fx", "fy", "cx", "cy"}


def test_missing_video_404(client):
    assert client.get("/api/videos/999/poses").status_code == 404


def test_anchors_get_and_post(client):
    anchors = client.get("/api/anchors").json()
    assert len(anchors) == 4
    created = client.post(
        "/api/anchors", json={"label": "T", "x": 1, "y": 2, "z": 3, "color": "#fff"}
    ).json()
    assert created["label"] == "T" and created["id"] > 0


def test_events(client):
    events = client.get("/api/events").json()
    assert len(events) == len(__import__("festival4d.synthetic", fromlist=["x"]).SEED_EVENTS)
    assert {"id", "t_global_s", "event_type", "source", "description"} <= set(events[0])


def test_pointcloud_is_ply(client):
    resp = client.get("/api/pointcloud")
    assert resp.status_code == 200
    assert resp.content[:3] == b"ply"


def test_audio_track_extracted_and_cached(client):
    """GET /api/audio/{id}: lazily-extracted AAC for the WebAudio master clock."""
    resp = client.get("/api/audio/1")
    assert resp.status_code == 200
    assert resp.headers["content-type"] == "audio/mp4"
    assert len(resp.content) > 1000
    # ftyp box near the start marks an MP4/M4A container.
    assert b"ftyp" in resp.content[:64]
    # Second request serves the cached file byte-identical (no re-extraction drift).
    assert client.get("/api/audio/1").content == resp.content
    assert client.get("/api/audio/999").status_code == 404


def test_video_range_returns_206(client):
    """Spec pitfall #3: <video> seeking needs HTTP Range -> 206."""
    resp = client.get("/media/cam0.mp4", headers={"Range": "bytes=0-100"})
    assert resp.status_code == 206
    assert resp.headers["content-range"].startswith("bytes 0-100/")
    assert len(resp.content) == 101


def test_cors_allows_vite_origin(client):
    resp = client.get("/api/manifest", headers={"Origin": "http://localhost:5173"})
    assert resp.headers.get("access-control-allow-origin") == "http://localhost:5173"


def test_detect_events_endpoint(client):
    # Lane D (M7) has landed: POST /api/events/detect runs real detection and returns
    # {result, events} (no more stub "note"). result is the run_events summary.
    resp = client.post("/api/events/detect", json={})
    assert resp.status_code == 200
    data = resp.json()
    assert "result" in data and "events" in data
    assert {"candidates", "classified", "candidates_only", "classifier_configured"} <= set(data["result"])
    assert isinstance(data["events"], list)
    assert client.post("/api/events/detect", json={}).status_code == 200  # idempotent


def test_annotation_resolves_to_anchor(client):
    # M8 (integration): a stored annotation now resolves to a 3D anchor. Single-view (no
    # event_id) falls back to the nearest point-cloud point / centroid-depth ray, so an
    # anchor + point come back. (Foundation asserted the stub contract anchor_id is None;
    # superseded by M8 landing — same situation as CR-1.)
    data = client.post(
        "/api/annotations",
        json={"video_id": 1, "t_video_s": 0.5, "bbox": [0.4, 0.4, 0.6, 0.6]},
    ).json()
    assert data["annotation_id"] > 0
    assert data["anchor_id"] is not None
    assert data["point"] is not None and len(data["point"]) == 3
    assert data["method"] in {"nearest_point", "ray_depth", "triangulated"}
    assert data["anchor"]["id"] == data["anchor_id"]


def test_event_patch_sets_user_source(client):
    events = client.get("/api/events").json()
    eid = events[0]["id"]
    patched = client.patch(f"/api/events/{eid}", json={"event_type": "pyro"}).json()
    assert patched["event_type"] == "pyro"
    assert patched["source"] == "user"  # user correction attribution
    assert client.patch("/api/events/99999", json={"event_type": "x"}).status_code == 404