5c2d7c6ea3
Fills events_ai.py against the frozen M7 contract:
- detect_candidates: RMS x spectral-flux, local-max over a 10s neighborhood,
>=P90, with an AC-RMS (std) silence/DC floor so flat audio yields no events.
Matches the fixture ground-truth pulses [3,10,17] within ~0.02s (<0.5s).
- GeminiClassifier (native video), ClaudeClassifier (claude-opus-4-8,
messages.parse, adaptive thinking, no temperature), LocalClassifier
(OpenAI-compatible, 6 frames, 1 retry). SDKs imported lazily.
- get_classifier: FESTIVAL4D_CLASSIFIER selection; unconfigured -> None
(candidates-only, logged).
- prepare_inputs: ffmpeg 3s clip (<=720p, audio kept) + 6 JPEGs (<=768px).
- run_events: reference-audio detect -> classify (insert-once in final state,
since db.py has no update-event helper); per-candidate exception isolation;
whole-track replaces machine events (keeps user edits), windowed is additive.
test_events_ai.py: candidate-vs-ground-truth (no ffmpeg/API), stub-classifier
orchestration, exception isolation, degradation, provider selection, silence.
Also updates the now-obsolete api stub test (test_detect_events_endpoint) to the
real M3 {result, events} shape; frozen api.py unchanged (see CR-1). pytest: 90 passed.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
113 lines
3.9 KiB
Python
113 lines
3.9 KiB
Python
"""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()
|
|
assert set(data) == {"videos", "t_global_max", "has_poses"}
|
|
assert data["has_poses"] is True
|
|
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_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_stored(client):
|
|
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 None # M8 resolution is integration work
|