cf6a6fb2c8
What now works: - M0 scaffold: pyproject (all spec deps), uv/py3.12 env, `python -m festival4d` CLI registering synthetic|ingest|sync|reconstruct|events|serve. Vite hello page. - Synthetic fixture (synthetic.py): 3 shifted-audio videos (offsets 0/+1370/-842 ms), camera-arc poses, stage point cloud -> points.ply, seeded events + anchors, ground_truth.json. `python -m festival4d synthetic` populates data/ + DB. - DB schema exactly per spec §2 (db.py) + CRUD helpers all lanes use. - M3 API (api.py) full against synthetic data: manifest/poses/pointcloud/anchors/ events/detect/annotations; Range-capable video serving (206 verified); CORS for any localhost origin. - Frozen geometry contract: geometry.colmap_to_threejs (M5 math) + unit test (3 known vectors, random round-trip, scipy oracle); mirrored frontend/src/lib/pose.js with identical POSE_TEST_VECTORS. Lane-B stubs: slerp_pose, ray_from_pixel, triangulate_rays, nearest_point_on_ray. - Classifier contract (events_ai.py): MomentClassification model + MomentClassifier protocol + Gemini/Claude/Local provider stubs. - Lane-owned modules stubbed with final signatures (ingest, audio_sync, frames, sfm, events_ai); cli/api catch NotImplementedError and degrade gracefully. - plan/CHANGE_REQUESTS.md created; plan/status/foundation.md updated. Acceptance: pytest 24 passed; serve endpoints verified via curl + browser (video seek, manifest fetch cross-origin, pose.js self-test, 0 console errors). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
107 lines
3.6 KiB
Python
107 lines
3.6 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_degrades_gracefully(client):
|
|
data = client.post("/api/events/detect", json={}).json()
|
|
assert "note" in data and "events" in data # stubbed lane D -> 200 with note
|
|
assert client.post("/api/events/detect", json={}).status_code == 200
|
|
|
|
|
|
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
|