5e2094f2ad
FastAPI service in infra/media-worker implementing every MediaInferenceProvider operation behind HTTP API v1, plus capabilities. Bearer auth, MEDIA_ROOT realpath enforcement, strict request schemas, cancellation on disconnect, GPU/CPU gates with 503 overflow, input size/duration limits, lazy SHA256-pinned model downloads, fail-closed redaction with atomic output, offline pytest suite with fake backends, CPU/CUDA/ROCm Dockerfile and a license review of every model. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0138YqDj3ttfayRwuMKKrqVj
89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
"""Operations without model weights, run for real on synthetic media."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.media_io import parse_ebur128
|
|
|
|
from .conftest import AUTH
|
|
|
|
|
|
def test_audio_features_on_click_track(client: TestClient, media_root: Path) -> None:
|
|
response = client.post("/v1/audioFeatures", headers=AUTH, json={"path": f"{media_root}/clicks.wav"})
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["integratedLufs"] is not None and -60 < body["integratedLufs"] < 0
|
|
assert body["truePeakDb"] is not None and body["truePeakDb"] <= 1.0
|
|
tempo = body["tempoBpm"]
|
|
assert tempo is not None and any(abs(tempo - bpm) < 6 for bpm in (60, 120, 240))
|
|
beats = body["beatsMs"]
|
|
assert len(beats) >= 6
|
|
gaps = np.diff(beats)
|
|
assert np.median(gaps) % 500 < 40 or 500 - np.median(gaps) % 500 < 40
|
|
# The trailing 1.5 s are silent.
|
|
assert any(7400 <= s["startMs"] <= 7700 and s["endMs"] >= 9400 for s in body["silences"]), body[
|
|
"silences"
|
|
]
|
|
energy = body["energy"]
|
|
assert energy[0]["tMs"] == 0 and all(0.0 <= e["value"] <= 1.0 for e in energy)
|
|
assert max(e["value"] for e in energy) == 1.0
|
|
assert all(isinstance(t, int) for t in beats)
|
|
|
|
|
|
def test_audio_features_without_audio_stream(client: TestClient, media_root: Path) -> None:
|
|
response = client.post("/v1/audioFeatures", headers=AUTH, json={"path": f"{media_root}/disc.mp4"})
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"integratedLufs": None,
|
|
"truePeakDb": None,
|
|
"tempoBpm": None,
|
|
"beatsMs": [],
|
|
"energy": [],
|
|
"silences": [],
|
|
}
|
|
|
|
|
|
def test_parse_ebur128_summary() -> None:
|
|
stderr = """
|
|
[Parsed_ebur128_0 @ 0x0] Summary:
|
|
|
|
Integrated loudness:
|
|
I: -23.4 LUFS
|
|
Threshold: -33.6 LUFS
|
|
|
|
True peak:
|
|
Peak: -1.2 dBFS
|
|
"""
|
|
assert parse_ebur128(stderr) == (-23.4, -1.2)
|
|
assert parse_ebur128("Summary:\n I: -inf LUFS\n Peak: -inf dBFS") == (None, None)
|
|
|
|
|
|
def test_scene_detection_finds_hard_cut(client: TestClient, media_root: Path) -> None:
|
|
response = client.post("/v1/detectScenes", headers=AUTH, json={"path": f"{media_root}/cut.mp4"})
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
shots = body["shots"]
|
|
assert len(shots) == 2, shots
|
|
assert abs(shots[1]["startMs"] - 2000) <= 80
|
|
assert shots[0]["startMs"] == 0 and abs(shots[-1]["endMs"] - 4000) <= 80
|
|
assert all(0.0 <= s["score"] <= 1.0 for s in shots)
|
|
assert body["scenes"][0]["startMs"] == 0
|
|
assert body["scenes"][-1]["endMs"] == shots[-1]["endMs"]
|
|
|
|
|
|
def test_saliency_points_at_bright_spot(client: TestClient, media_root: Path) -> None:
|
|
response = client.post(
|
|
"/v1/saliency", headers=AUTH, json={"path": f"{media_root}/spot.mp4", "timesMs": [0, 200]}
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
points = response.json()["points"]
|
|
assert [p["tMs"] for p in points] == [0, 200]
|
|
for point in points:
|
|
assert point["cx"] > 0.6
|
|
assert 0.3 < point["cy"] < 0.7
|
|
assert 0.0 < point["strength"] <= 1.0
|