36bacf67f0
Records rather than streams: each device records a clip WITH AUDIO and uploads
it; the normal offline pipeline takes over. Fits the design almost for free —
MediaRecorder's webm/mp4 are already ingestible, and audio-based sync means
devices need no clock sync and needn't start together.
- capture.py: POST /api/capture/upload (streamed, sanitized name, size cap,
partial cleanup), heartbeat/devices registry, GET /capture page.
- static/capture.html: standalone page (no build step) — device picker, live
preview, record->upload, 1fps snapshot heartbeat, 'who's shooting what'
monitor, and secure-context detection that tells you to use
(getUserMedia needs HTTPS — the #1 gotcha for phones).
- OPT-IN via FESTIVAL4D_CAPTURE=1; manifest reports has_capture and the viewer
only then shows a Capture link. Public deploy leaves it off (open upload
endpoint would be unsafe) — noted in DEPLOY.md.
- Adds python-multipart dep. Suite 103 -> 119.
Also fixes test_manifest_shape, which was already RED on main: f12b6e0 added
has_splat to the manifest without updating the exact-set lock. Updated to the
true set (now also has_capture); the lock is what caught both drifts.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
132 lines
5.1 KiB
Python
132 lines
5.1 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()
|
|
# 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_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
|