#!/usr/bin/env python3 """M1 smoke test for scenegod.server — stdlib only, real uvicorn on a test port. python3 scripts/test_server.py # green = M1 code-done Spins up the server against throwaway asset/scene dirs, then asserts tree grouping, file streaming, path-traversal rejection, scene round-trip, and validation failures. """ import base64 import json import os import re import shutil import subprocess import sys import tempfile import threading import time import urllib.error import urllib.parse import urllib.request from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path ROOT = Path(__file__).resolve().parent.parent PY = ROOT / ".venv" / "bin" / "python" PORT = 8099 BASE = f"http://127.0.0.1:{PORT}" def req(method, path, body=None): data = body.encode() if isinstance(body, str) else body r = urllib.request.Request(BASE + path, data=data, method=method) try: with urllib.request.urlopen(r) as resp: return resp.status, resp.read() except urllib.error.HTTPError as e: return e.code, e.read() def get_json(path): st, b = req("GET", path) assert st == 200, f"{path} -> {st}" return json.loads(b) def ff(*args): subprocess.run(["ffmpeg", "-y", *args], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def probe_dur(f): return float(subprocess.check_output( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nokey=1", str(f)], text=True).strip()) def seg_db(f, ss, dur): """mean volume (dB) of one window of a media file — digital silence reads about -91 dB, so this is how a test asks 'is the audio actually HERE?'. Refuses to answer when the window decoded NO samples: our muxes apad to the video length, so an empty window means a broken stream, not a quiet one (an atrim+apad graph once produced exactly that — aac stream, zero audio).""" p = subprocess.run(["ffmpeg", "-v", "info", "-ss", str(ss), "-t", str(dur), "-i", str(f), "-af", "volumedetect", "-f", "null", "-"], capture_output=True, text=True) m = re.search(r"mean_volume:\s*(-?[\d.]+) dB", p.stderr) if not m: raise AssertionError(f"no audio samples in {Path(f).name} at {ss}..{ss+dur}s\n" + p.stderr[-400:]) return float(m.group(1)) def main(): tmp = Path(tempfile.mkdtemp(prefix="scenegod_test_")) assets, scenes = tmp / "assets", tmp / "scenes" # lady.glb + lady.fbx + lady.jpg share a stem -> one entry, thumb split off (assets / "characters" / "pack01").mkdir(parents=True) for f in ("lady.glb", "lady.fbx", "lady.jpg"): (assets / "characters" / "pack01" / f).write_bytes(b"x") (assets / "backdrops").mkdir(parents=True) (assets / "backdrops" / "street.jpg").write_bytes(b"img") for d in ("animations", "props", "audio"): (assets / d).mkdir() (assets / "props" / "box.glb").write_bytes(b"b") # no image sidecar renders, sequences, films = tmp / "renders", tmp / "sequences", tmp / "films" env = {**os.environ, "SCENEGOD_ASSETS": str(assets), "SCENEGOD_SCENES": str(scenes), "SCENEGOD_RENDERS": str(renders), "SCENEGOD_SEQUENCES": str(sequences), "SCENEGOD_FILMS": str(films)} env.pop("SCENEGOD_LLM_URL", None) # main instance must 503 on /director proc = subprocess.Popen( [str(PY), "-m", "uvicorn", "scenegod.server:app", "--port", str(PORT)], cwd=ROOT, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: for _ in range(50): # wait for boot try: req("GET", "/scenes") break except urllib.error.URLError: time.sleep(0.2) else: raise SystemExit("server did not start") n = 0 # --- tree grouping --- tree = get_json("/assets/tree") chars = tree["characters"] assert len(chars) == 1 and chars[0]["name"] == "lady", chars assert chars[0]["path"] == "characters/pack01/lady.glb", chars # glb preferred assert set(chars[0]["formats"]) == {"glb", "fbx"}, chars # jpg not a format assert chars[0]["thumb"] == "characters/pack01/lady.jpg", chars assert tree["backdrops"][0]["path"] == "backdrops/street.jpg", tree["backdrops"] n += 1 # --- file streaming + thumbnail sidecar --- st, b = req("GET", "/assets/file?path=backdrops/street.jpg") assert st == 200 and b == b"img", (st, b) st, b = req("GET", "/assets/thumb?path=" + urllib.parse.quote("characters/pack01/lady.glb")) assert st == 200 and b == b"x", (st, b) # resolves lady.jpg sidecar assert req("GET", "/assets/thumb?path=props/box.glb")[0] == 404 # no sidecar n += 1 # --- path traversal rejected --- for bad in ("../../etc/hosts", "/etc/hosts", "..%2f..%2fetc%2fhosts", "characters/../../../etc/hosts"): st, _ = req("GET", "/assets/file?path=" + bad) assert st >= 400, f"traversal {bad!r} not rejected: {st}" n += 1 # --- scene round-trip --- scene = {"version": 1, "name": "s1", "fps": 30, "duration": 5.0, "entities": [{"id": "e1", "kind": "camera", "label": "cam", "tracks": {"transform": [{"t": 0}, {"t": 1}]}}], "cameraCuts": [{"t": 0, "camera": "e1"}]} st, _ = req("POST", "/scenes/" + urllib.parse.quote("My Scene!"), json.dumps(scene)) assert st == 200, st assert get_json("/scenes/my-scene")["duration"] == 5.0 # slugified assert any(s["name"] == "my-scene" for s in get_json("/scenes")) n += 1 # --- validation failures --- bad_scenes = [ {"version": 2, "entities": []}, # bad version {"version": 1, "entities": [{"id": "a"}, {"id": "a"}]}, # dup id {"version": 1, "entities": [{"id": "a", "tracks": # unsorted {"transform": [{"t": 1}, {"t": 0}]}}]}, {"version": 1, "entities": [], "cameraCuts": [{"camera": "nope"}]}, # bad cut {"version": 1, "entities": [{"id": "a", "tracks": {"clips": [ # overlap {"start": 0, "in": 0, "out": 2, "loop": 1, "fade": 0}, {"start": 1, "in": 0, "out": 2}]}}]}, ] for bs in bad_scenes: st, b = req("POST", "/scenes/bad", json.dumps(bs)) assert st == 422, f"expected 422 for {bs}, got {st}" assert json.loads(b)["errors"], b # a fade-covered overlap is allowed ok = {"version": 1, "entities": [{"id": "a", "tracks": {"clips": [ {"start": 0, "in": 0, "out": 2, "loop": 1, "fade": 0.5}, {"start": 1.6, "in": 0, "out": 2}]}}]} st, _ = req("POST", "/scenes/ok-fade", json.dumps(ok)) assert st == 200, st n += 1 # --- starter templates (M7-C): listed, read-only, guarded, and NOT ROTTEN --- tpl = get_json("/templates") assert sorted(t["name"] for t in tpl) == [ "empty-stage", "record-store", "two-hander-golden-hour"], tpl assert tpl[0]["name"] == "two-hander-golden-hour", "order: lead with the wow one" for t in tpl: assert t["title"] and t["description"], t # traversal / absolute / URL-encoded .. / unknown name for bad in ("../../etc/hosts", "..%2f..%2fetc%2fhosts", "%2e%2e%2f%2e%2e%2fetc%2fhosts", urllib.parse.quote("/etc/hosts", safe=""), "nope"): st, _ = req("GET", "/templates/" + bad) assert st >= 400, f"template path {bad!r} not rejected: {st}" # read-only: no write verbs assert req("POST", "/templates/empty-stage", "{}")[0] in (404, 405) assert req("DELETE", "/templates/empty-stage")[0] in (404, 405) real_assets = ROOT / "assets" for t in tpl: js = get_json("/templates/" + t["name"]) assert js["version"] == 1 and js["name"] == t["name"], js["name"] # round-trips through the REAL validator: save it as a scene like the UI does st, b = req("POST", "/scenes/tplcheck-" + t["name"], json.dumps(js)) assert st == 200, f"template {t['name']} fails validate_scene: {b}" refs = [t["thumb"]] if t.get("thumb") else [] for e in js["entities"]: src = e.get("source") or {} if e["kind"] in ("camera", "light"): assert src.get("type") == "none", \ f"{t['name']}/{e['id']}: {e['kind']} must have source type 'none'" if src.get("type") == "assets": refs.append(src["path"]) p = e.get("params") or {} refs += [p[k] for k in ("video", "image") if p.get(k)] refs += [c["path"] for c in (e.get("tracks") or {}).get("clips") or []] refs += [a["path"] for a in js.get("audio") or []] if real_assets.is_dir(): # the check that stops a rotten template shipping for r in refs: assert (real_assets / r).is_file(), \ f"template {t['name']}: missing asset {r!r}" assert refs or t["name"] == "empty-stage", f"{t['name']} references nothing" if not real_assets.is_dir(): print("(template asset-existence check skipped — no assets/ dir)") n += 1 # --- render pipeline: begin -> post 30 PNGs -> end -> poll -> mp4 --- if shutil.which("ffmpeg") and shutil.which("ffprobe"): fr = tmp / "genframes" fr.mkdir() subprocess.run(["ffmpeg", "-y", "-f", "lavfi", "-i", "color=c=red:s=64x64", "-frames:v", "30", str(fr / "%06d.png")], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) rid = json.loads(req("POST", "/render/begin", json.dumps({"name": "t", "fps": 30}))[1])["renderId"] for i, png in enumerate(sorted(fr.glob("*.png"))): st, _ = req("POST", f"/render/{rid}/frame/{i}", png.read_bytes()) assert st == 200, st # M4: mux a tone from assets/audio/test/ via the scene audio[] contract (assets / "audio" / "test").mkdir(parents=True) subprocess.run(["ffmpeg", "-y", "-f", "lavfi", "-i", "sine=frequency=440:duration=1", str(assets / "audio" / "test" / "tone.wav")], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) end_body = json.dumps({"audio": [{"path": "audio/test/tone.wav", "start": 0, "gain": 1.0}]}) assert req("POST", f"/render/{rid}/end", end_body)[0] == 200 for _ in range(100): # poll up to ~20s state = json.loads(req("GET", f"/render/{rid}/status")[1])["state"] if state in ("done", "error"): break time.sleep(0.2) assert state == "done", state st, mp4 = req("GET", f"/render/{rid}/out.mp4") assert st == 200 and mp4[:4] != b"", st out = str(renders / rid / "out.mp4") dur = subprocess.check_output( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nokey=1", out], text=True).strip() assert abs(float(dur) - 1.0) < 0.2, f"mp4 duration {dur}s, expected ~1s" codecs = subprocess.check_output( ["ffprobe", "-v", "error", "-select_streams", "a", "-show_entries", "stream=codec_type", "-of", "default=nw=1:nokey=1", out], text=True).strip() assert "audio" in codecs, f"no audio stream muxed: {codecs!r}" # bad audio path rejected before encode assert req("POST", f"/render/{rid}/end", json.dumps({"audio": [{"path": "audio/nope.wav"}]}))[0] == 400 n += 1 # --- M9-C1: audio in/out honoured at the mux ------------------- # trim.wav = 1s of silence, then 1s of 880Hz. Trimming to the tone # and starting at 0.5s must put SOUND at 0.5-1.5s and nothing else; # the same clip untrimmed must still lead with its silent second. ff("-f", "lavfi", "-i", "aevalsrc='if(gte(t,1),0.7*sin(2*PI*880*t),0)':s=44100:d=2", "-c:a", "pcm_s16le", str(assets / "audio" / "test" / "trim.wav")) pngs = sorted(fr.glob("*.png")) def render_audio(audio, frames=90): # 3s of picture @30fps r2 = json.loads(req("POST", "/render/begin", json.dumps({"name": "t", "fps": 30}))[1])["renderId"] for i in range(frames): assert req("POST", f"/render/{r2}/frame/{i}", pngs[i % len(pngs)].read_bytes())[0] == 200 assert req("POST", f"/render/{r2}/end", json.dumps({"audio": audio}))[0] == 200 for _ in range(150): s2 = json.loads(req("GET", f"/render/{r2}/status")[1])["state"] if s2 in ("done", "error"): break time.sleep(0.2) assert s2 == "done", s2 return renders / r2 / "out.mp4" cut = render_audio([{"path": "audio/test/trim.wav", "start": 0.5, "in": 1.0, "out": 2.0, "gain": 1.0}]) assert abs(probe_dur(cut) - 3.0) < 0.2, probe_dur(cut) assert seg_db(cut, 0.0, 0.4) < -50, f"trimmed clip leaks before start: {seg_db(cut,0,0.4)}" assert seg_db(cut, 0.6, 0.8) > -35, f"trimmed tone missing at 0.6-1.4s: {seg_db(cut,0.6,0.8)}" assert seg_db(cut, 1.7, 1.0) < -50, f"trimmed clip runs past out: {seg_db(cut,1.7,1.0)}" whole = render_audio([{"path": "audio/test/trim.wav", "start": 0.5, "gain": 1.0}]) assert seg_db(whole, 0.6, 0.8) < -50, "untrimmed clip must still play its silent head" assert seg_db(whole, 1.6, 0.8) > -35, "untrimmed tone should land a second later" # and the trim is validated, not trusted for bad in ({"in": -1}, {"in": 2, "out": 1}, {"in": "x"}, {"start": -1}): assert req("POST", f"/render/{rid}/end", json.dumps( {"audio": [{"path": "audio/test/trim.wav", **bad}]}))[0] == 400, bad # a window past EOF would fail the MUX (ffmpeg -22) — caught up front st, b = req("POST", f"/render/{rid}/end", json.dumps( {"audio": [{"path": "audio/test/trim.wav", "in": 99}]})) assert st == 400 and b"past the end" in b, b # a tail past EOF is just no tail trim: renders, and still plays past = render_audio([{"path": "audio/test/trim.wav", "start": 0.0, "in": 1.0, "out": 60.0, "gain": 1.0}]) assert seg_db(past, 0.1, 0.8) > -35, "clamped tail trim lost the audio" n += 1 else: print("(render test skipped — ffmpeg/ffprobe not found)") # --- sequences (M8-C): CRUD round-trip --- for nm, dur, fps in (("clip-a", 2.0, 30), ("clip-b", 3.0, 30), ("clip-24", 2.0, 24)): st, _ = req("POST", "/scenes/" + nm, json.dumps( {"version": 1, "name": nm, "fps": fps, "duration": dur, "entities": []})) assert st == 200, (nm, st) seq = {"version": 1, "name": "my-film", "title": "My Film", "shots": [{"scene": "clip-a", "in": 0.0, "out": 2.0, "transition": "dissolve", "transitionDur": 0.5}, {"scene": "clip-b", "in": 0.5, "out": 3.0}]} st, b = req("POST", "/sequences/" + urllib.parse.quote("My Film!"), json.dumps(seq)) assert st == 200, b saved = json.loads(b) assert saved["name"] == "my-film", saved # slugified assert abs(saved["duration"] - (2.0 + 2.5 - 0.5)) < 1e-6, saved # dissolve eats 0.5s assert get_json("/sequences/my-film") == seq, "sequence round-trip must be lossless" row = [s for s in get_json("/sequences") if s["name"] == "my-film"][0] assert row["title"] == "My Film" and row["shots"] == 2 and row["ok"], row assert req("DELETE", "/sequences/my-film")[0] == 200 assert req("GET", "/sequences/my-film")[0] == 404 assert req("DELETE", "/sequences/my-film")[0] == 404 n += 1 # --- sequence validation: every violation a 422 that says what's wrong --- def sq(shots, **kw): return json.dumps({"version": 1, "shots": shots, **kw}) bad_seqs = [ (sq([{"scene": "nope", "in": 0, "out": 1}]), "unknown scene"), (sq([{"scene": "clip-a", "in": 2.0, "out": 1.0}]), "less than"), (sq([{"scene": "clip-a", "in": 0, "out": 1, "transition": "wipe"}]), "transition"), (sq([]), "non-empty"), (json.dumps({"version": 2, "shots": [{"scene": "clip-a"}]}), "version"), (sq([{"scene": "clip-a", "in": 0, "out": 9}]), "past the end"), (sq([{"scene": "clip-a", "transition": "dissolve", "transitionDur": 5}, {"scene": "clip-b"}]), "shorter than"), (sq([{"scene": "clip-a", "transition": "dissolve", "transitionDur": 0}]), "needs"), (sq([{"scene": "clip-a", "transitionDur": 99, "transition": "dissolve"}]), "transitionDur"), (sq([{"scene": "clip-a"}, {"scene": "clip-24"}]), "one fps"), (sq([{"scene": "../../etc/hosts"}]), "bad scene name"), (sq([{"scene": "clip-a"}], audio=[{"path": "../../etc/hosts"}]), "bad path"), (sq([{"scene": "clip-a"}], audio=[{"path": "audio/nope.wav"}]), "not found"), ] for body, needle in bad_seqs: st, b = req("POST", "/sequences/bad", body) assert st == 422, f"expected 422 for {body}, got {st}" errs = json.loads(b)["errors"] assert any(needle in e for e in errs), (needle, errs) assert req("POST", "/sequences/bad", "{not json")[0] == 400 # name guards on read/delete (same discipline as /scenes, /templates) for bad in ("../../etc/hosts", "..%2f..%2fetc%2fhosts", urllib.parse.quote("/etc/hosts", safe=""), "nope"): assert req("GET", "/sequences/" + bad)[0] >= 400, bad assert req("DELETE", "/sequences/" + bad)[0] >= 400, bad n += 1 # --- FILM smoke: two shots -> one mp4, cut and dissolve (M8-C) --- if shutil.which("ffmpeg") and shutil.which("ffprobe"): def fake_shot_render(dur, w=64, h=64, fps=30, color="red"): """A *finished* /render session minus the browser: lavfi mp4 + meta.""" rid = os.urandom(16).hex() d = renders / rid (d / "frames").mkdir(parents=True) (d / "meta.json").write_text(json.dumps( {"name": "shot", "fps": fps, "width": w, "height": h})) subprocess.run(["ffmpeg", "-y", "-f", "lavfi", "-i", f"color=c={color}:s={w}x{h}:r={fps}:d={dur}", "-c:v", "libx264", "-pix_fmt", "yuv420p", str(d / "out.mp4")], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return rid def probe(f, entries="format=duration"): return subprocess.check_output( ["ffprobe", "-v", "error", "-show_entries", entries, "-of", "default=nw=1:nokey=1", str(f)], text=True).strip() def make_film(seq_name, body): assert req("POST", "/sequences/" + seq_name, json.dumps(body))[0] == 200, seq_name st, b = req("POST", f"/sequences/{seq_name}/render", json.dumps({"width": 64, "height": 64})) assert st == 200, b return json.loads(b) def finish_film(fid, rids): for i, rid in enumerate(rids): st, b = req("POST", f"/films/{fid}/shot/{i}", json.dumps({"renderId": rid})) assert st == 200, b assert req("POST", f"/films/{fid}/end", "{}")[0] == 200 for _ in range(150): # poll up to ~30s stt = json.loads(req("GET", f"/films/{fid}/status")[1]) if stt["state"] in ("done", "error"): break time.sleep(0.2) assert stt["state"] == "done", stt assert stt["shot"] == stt["of"] == len(rids), stt assert req("GET", f"/films/{fid}/out.mp4")[0] == 200 return films / fid / "out.mp4" shots = [{"scene": "clip-a", "in": 0.0, "out": 2.0}, {"scene": "clip-b", "in": 0.0, "out": 3.0}] film = make_film("film-cut", {"version": 1, "shots": shots}) assert film["fps"] == 30 and film["duration"] == 5.0, film assert [s["scene"] for s in film["shots"]] == ["clip-a", "clip-b"], film fid = film["filmId"] assert json.loads(req("GET", f"/films/{fid}/status")[1]) == { "state": "collecting", "shot": 0, "of": 2, "log": ""} assert req("POST", f"/films/{fid}/end", "{}")[0] == 400 # shots not registered yet rid_a, rid_b = fake_shot_render(2.0), fake_shot_render(3.0, color="blue") # guards: bad index, bad/unknown renderId, wrong resolution, unknown film assert req("POST", f"/films/{fid}/shot/9", json.dumps({"renderId": rid_a}))[0] == 400 assert req("POST", f"/films/{fid}/shot/0", json.dumps({"renderId": "../../etc"}))[0] == 400 assert req("POST", f"/films/{fid}/shot/0", json.dumps({"renderId": "deadbeefdeadbeef"}))[0] == 400 # no out.mp4 wrong = fake_shot_render(1.0, w=32, h=32) st, b = req("POST", f"/films/{fid}/shot/0", json.dumps({"renderId": wrong})) assert st == 400 and b"one resolution per film" in b, b assert req("GET", "/films/nosuchfilm/status")[0] == 404 out = finish_film(fid, [rid_a, rid_b]) dur = float(probe(out)) assert abs(dur - 5.0) < 0.15, f"cut film {dur}s, expected ~5s (2+3)" assert probe(out, "stream=width,height").split() == ["64", "64"], probe(out, "stream=width,height") n += 1 # dissolve variant: sum MINUS the transition, plus a film-level audio bed tone = assets / "audio" / "test" / "tone.wav" if not tone.is_file(): tone.parent.mkdir(parents=True, exist_ok=True) subprocess.run(["ffmpeg", "-y", "-f", "lavfi", "-i", "sine=frequency=440:duration=1", str(tone)], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) dshots = [dict(shots[0], transition="dissolve", transitionDur=0.5), shots[1]] # M9-C1: the film-level bed is trimmed too — trim.wav's tone (source # 1..2s) placed at 2.0s must be silent before it and present after. film2 = make_film("film-dissolve", {"version": 1, "shots": dshots, "audio": [{"path": "audio/test/trim.wav", "start": 2.0, "in": 1.0, "out": 2.0, "gain": 0.8}]}) assert film2["duration"] == 4.5, film2 assert not film2["warnings"], film2 # bed present -> no warning out2 = finish_film(film2["filmId"], [fake_shot_render(2.0), fake_shot_render(3.0, color="blue")]) dur2 = float(probe(out2)) assert abs(dur2 - 4.5) < 0.15, f"dissolve film {dur2}s, expected ~4.5s (2+3-0.5)" codecs = probe(out2, "stream=codec_type").split() assert "audio" in codecs, f"film audio bed not muxed: {codecs}" assert seg_db(out2, 0.0, 1.8) < -50, f"film bed leaks before start: {seg_db(out2,0,1.8)}" assert seg_db(out2, 2.1, 0.8) > -35, f"film bed tone missing: {seg_db(out2,2.1,0.8)}" # a bad trim is a 422 with a message, same as every other violation st, b = req("POST", "/sequences/bad", json.dumps( {"version": 1, "shots": dshots, "audio": [{"path": "audio/test/trim.wav", "in": 3, "out": 1}]})) assert st == 422 and b"less than" in b, b n += 1 else: print("(film test skipped — ffmpeg/ffprobe not found)") # --- shipped example sequence must not rot (same rule as templates) --- real_seqs, real_scenes = ROOT / "sequences", ROOT / "scenes" shipped = sorted(real_seqs.glob("*.json")) if real_seqs.is_dir() else [] assert shipped, "no example sequence shipped in sequences/" for p in shipped: js = json.loads(p.read_text()) assert js["version"] == 1 and js["shots"], p.name if real_scenes.is_dir(): for sh in js["shots"]: sc = real_scenes / (sh["scene"] + ".json") assert sc.is_file(), f"{p.name}: missing scene {sh['scene']!r}" dur = json.loads(sc.read_text()).get("duration", 0) assert sh.get("out", dur) <= dur + 1e-6, f"{p.name}: shot past end of {sh['scene']}" n += 1 # --- M9-C2: BEAT DETECTION on click tracks of KNOWN tempo ------------ if shutil.which("ffmpeg"): (assets / "audio" / "test").mkdir(parents=True, exist_ok=True) def click(name, bpm, dur=20): """Exactly-periodic click track: mod(t, period) in aevalsrc is double-precision, so beat k is at exactly k*period (unlike a concatenated or gated source, which drifts).""" per = 60.0 / bpm ff("-f", "lavfi", "-i", f"aevalsrc='0.8*sin(2*PI*1200*t)" f"*exp(-45*mod(t,{per:.9f}))':s=44100:d={dur}", "-c:a", "pcm_s16le", str(assets / "audio" / "test" / name)) return per # 118 @ 60s is not padding: harmonics falling off the end of the lag # table used to penalise their own lag, and the parabola was then # fitted to a point on a slope — together they read this one as 114.8. for name, bpm, dur in (("click120.wav", 120, 20), ("click128.wav", 128, 20), ("click118.wav", 118, 60)): per = click(name, bpm, dur=dur) r = get_json("/beats?path=audio/test/" + name) assert abs(r["bpm"] - bpm) <= 2, f"{name}: detected {r['bpm']}, truth {bpm}" assert r["confidence"] > 0.5, f"{name}: confidence {r['confidence']} on a click track" assert abs(r["duration"] - dur) < 0.2 and r["meter"] == 4, r assert len(r["beats"]) >= int(dur / per) - 1, len(r["beats"]) for i, b in enumerate(r["beats"][:12]): # grid lands on the real clicks assert abs(b - i * per) < 0.03, f"{name}: beat {i} at {b}s, truth {i*per:.4f}s" d = max(abs(b - round(b / per) * per) for b in r["beats"]) # whole file assert d < 0.020, f"{name}: grid slips {d*1000:.0f}ms somewhere in the file" assert r["downbeats"], r i0 = r["beats"].index(r["downbeats"][0]) # 4/4: every 4th beat assert [r["beats"][i0 + 4 * k] for k in range(3)] == r["downbeats"][:3], r["downbeats"] # ACCUMULATED DRIFT, not average tempo. A 0.3% tempo bias is invisible # in a 20s clip and is 0.6s of slip by the end of a 200s one — cuts # snapped late in a track would land visibly off the kick. per = click("long128.wav", 128, dur=200) r = get_json("/beats?path=audio/test/long128.wav") assert abs(r["bpm"] - 128) < 0.1, f"long file: {r['bpm']}" assert r["beats"][-1] > 195, f"grid stops at {r['beats'][-1]}s of a 200s file" drift = max(abs(b - round(b / per) * per) for b in r["beats"]) assert drift < 0.020, f"grid slips {drift*1000:.0f}ms across a 200s file" # The M9 bug this pins: a percussive attack is a HIGH-frequency # transient riding a slower low-frequency body. Decode at 22050 and # the transient is gone, so the onset lands on the body's energy peak # ~25ms late — and by a DIFFERENT amount per beat once the mix # evolves, which is a tempo error, not an offset. 16kHz tick over a # slow 60Hz body: at 44100 this reads +1ms, at 22050 +27ms. p124 = 60.0 / 124 ff("-f", "lavfi", "-i", f"aevalsrc='0.45*sin(2*PI*60*t)*(1-exp(-8*mod(t,{p124:.9f})))" f"*exp(-5*mod(t,{p124:.9f}))" f"+0.5*sin(2*PI*16000*t)*exp(-150*mod(t,{p124:.9f}))':s=44100:d=60", "-c:a", "pcm_s16le", str(assets / "audio" / "test" / "hftick.wav")) r = get_json("/beats?path=audio/test/hftick.wav") assert abs(r["bpm"] - 124) < 0.1, f"hf transient: {r['bpm']}" for i, b in enumerate(r["beats"][:12]): assert abs(b - i * p124) < 0.012, ( f"hftick beat {i} at {b}s, truth {i*p124:.4f}s — the decode is " f"throwing away the transient that defines the attack") # The two shipped fixtures are GROUND TRUTH: measured straight from # PCM, beat120's onsets sit on 0.5000s multiples to the microsecond # (120.003 BPM) and house128's on 0.468735s (128.004). assets/ is # gitignored, so skip when they are not on this box. for name, truth in (("beat120.wav", 120.003), ("house128.wav", 128.004)): srcf = ROOT / "assets" / "audio" / "test" / name if not srcf.is_file(): print(f"(fixture check skipped — no assets/audio/test/{name})") continue shutil.copy(srcf, assets / "audio" / "test" / name) r = get_json("/beats?path=audio/test/" + name) assert abs(r["bpm"] - truth) < 0.2, f"{name}: read {r['bpm']}, truth {truth}" p = 60.0 / truth d = max(abs(b - round(b / p) * p) for b in r["beats"]) assert d < 0.020, f"{name}: grid is {d*1000:.0f}ms off the real kick" # mush must LOOK like mush — low confidence, no crash, still 200 ff("-f", "lavfi", "-i", "anoisesrc=d=10:c=pink:a=0.3", "-c:a", "pcm_s16le", str(assets / "audio" / "test" / "noise.wav")) ff("-f", "lavfi", "-i", "anullsrc=r=44100:cl=mono", "-t", "8", "-c:a", "pcm_s16le", str(assets / "audio" / "test" / "quiet.wav")) for name in ("noise.wav", "quiet.wav"): r = get_json("/beats?path=audio/test/" + name) assert r["confidence"] < 0.35, f"{name}: confidence {r['confidence']} on non-music" assert isinstance(r["beats"], list) and isinstance(r["bpm"], (int, float)), r assert get_json("/beats?path=audio/test/quiet.wav")["beats"] == [], "silence has no grid" # guards + cache for bad in ("../../etc/hosts", "..%2f..%2fetc%2fhosts", urllib.parse.quote("/etc/hosts", safe="")): assert req("GET", "/beats?path=" + bad)[0] >= 400, bad assert req("GET", "/beats?path=audio/test/nope.wav")[0] == 404 click("cachecheck.wav", 124, dur=40) # never analysed yet: first call is cold t0 = time.time() a1 = get_json("/beats?path=audio/test/cachecheck.wav") t1 = time.time() a2 = get_json("/beats?path=audio/test/cachecheck.wav") t2 = time.time() assert abs(a1["bpm"] - 124) <= 2, a1["bpm"] assert a1 == a2, "cached /beats must return the same grid" assert (t2 - t1) < (t1 - t0) / 2, \ f"second /beats call not cached ({t2-t1:.3f}s vs {t1-t0:.3f}s)" n += 1 else: print("(beats test skipped — ffmpeg not found)") # --- the ffmpeg/rhubarb gates: a box without the binaries 503s cleanly --- port4 = PORT + 5 p4 = subprocess.Popen( [str(PY), "-m", "uvicorn", "scenegod.server:app", "--port", str(port4)], cwd=ROOT, env={**env, "PATH": "/usr/bin:/bin"}, # no homebrew -> no ffmpeg/rhubarb stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: for _ in range(50): try: urllib.request.urlopen(f"http://127.0.0.1:{port4}/scenes") break except urllib.error.URLError: time.sleep(0.2) for path in ("/beats?path=audio/test/tone.wav", "/rhubarb?path=audio/test/tone.wav"): try: urllib.request.urlopen(f"http://127.0.0.1:{port4}{path}") raise AssertionError(f"{path} should 503 without ffmpeg/rhubarb") except urllib.error.HTTPError as e: assert e.code == 503, (path, e.code) n += 1 finally: p4.terminate() p4.wait() # --- rhubarb: 503 when the binary is absent (happy path needs rhubarb) --- if not shutil.which("rhubarb"): assert req("GET", "/rhubarb?path=audio/x.wav")[0] == 503 n += 1 # --- MODELBEAST proxy: gating + two-step contract (mock MB, no farm jobs) --- assert req("POST", "/mb/submit", json.dumps({"operator": "tts"}))[0] == 503 # gated off n += 1 captured = [] class MB(BaseHTTPRequestHandler): def do_POST(self): data = self.rfile.read(int(self.headers.get("Content-Length", 0))) captured.append({"path": self.path, "auth": self.headers.get("Authorization"), "ctype": self.headers.get("Content-Type", ""), "body": data}) self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps( {"id": "asset123" if self.path == "/api/assets" else "job456"}).encode()) def log_message(self, *a): pass mbport, port2 = PORT + 1, PORT + 2 srv = HTTPServer(("127.0.0.1", mbport), MB) threading.Thread(target=srv.serve_forever, daemon=True).start() env2 = {**env, "SCENEGOD_MB": "1", "MB_HOST": f"http://127.0.0.1:{mbport}", "MB_TOKEN": "secret"} p2 = subprocess.Popen( [str(PY), "-m", "uvicorn", "scenegod.server:app", "--port", str(port2)], cwd=ROOT, env=env2, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: for _ in range(50): try: urllib.request.urlopen(f"http://127.0.0.1:{port2}/scenes") break except urllib.error.URLError: time.sleep(0.2) sub = {"operator": "voice-clone", "params": {"text": "hi"}, "file_b64": base64.b64encode(b"WAVDATA").decode(), "file_name": "in.wav"} r = urllib.request.urlopen(urllib.request.Request( f"http://127.0.0.1:{port2}/mb/submit", data=json.dumps(sub).encode(), method="POST", headers={"Content-Type": "application/json"})) out = json.loads(r.read()) assert out == {"job_id": "job456", "asset_id": "asset123"}, out assert [c["path"] for c in captured] == ["/api/assets", "/api/jobs"], captured assert all(c["auth"] == "Bearer secret" for c in captured), "token not forwarded" assert captured[0]["ctype"].startswith("multipart/form-data"), captured[0]["ctype"] assert b"WAVDATA" in captured[0]["body"], "file bytes not uploaded" job = json.loads(captured[1]["body"]) assert job["asset_id"] == "asset123" and job["operator"] == "voice-clone", job n += 1 finally: p2.terminate() p2.wait() srv.shutdown() # --- /director: 503 unconfigured; mock LLM -> ops validated, garbage rejected --- dbody = json.dumps({"text": "medium shot of lady, golden hour", "scene": {"entities": [{"id": "e1", "kind": "character", "label": "lady"}, {"id": "e7", "kind": "camera", "label": "cam A"}], "time": 2.5}}) assert req("POST", "/director", dbody)[0] == 503 # env unset -> hint, no crash llm_captured = [] canned = {"ops": [ {"op": "shot", "subject": "e1", "shot": "MS", "angle": "low", "focal": 35}, {"op": "light", "preset": "golden-hour"}, {"op": "mark", "subject": "e1", "mark": "walk-to-mark"}, {"op": "explode", "subject": "e1"}, # unknown op {"op": "shot", "subject": "ghost", "shot": "CU"}, # unknown entity id {"op": "light", "preset": "disco"}, # unknown preset ]} class LLM(BaseHTTPRequestHandler): def do_POST(self): llm_captured.append(json.loads( self.rfile.read(int(self.headers.get("Content-Length", 0))))) self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps( {"choices": [{"message": {"content": json.dumps(canned)}}]}).encode()) def log_message(self, *a): pass llmport, port3 = PORT + 3, PORT + 4 srv3 = HTTPServer(("127.0.0.1", llmport), LLM) threading.Thread(target=srv3.serve_forever, daemon=True).start() env3 = {**env, "SCENEGOD_LLM_URL": f"http://127.0.0.1:{llmport}", "SCENEGOD_LLM_MODEL": "mockmodel"} p3 = subprocess.Popen( [str(PY), "-m", "uvicorn", "scenegod.server:app", "--port", str(port3)], cwd=ROOT, env=env3, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: for _ in range(50): try: urllib.request.urlopen(f"http://127.0.0.1:{port3}/scenes") break except urllib.error.URLError: time.sleep(0.2) r = urllib.request.urlopen(urllib.request.Request( f"http://127.0.0.1:{port3}/director", data=dbody.encode(), method="POST", headers={"Content-Type": "application/json"})) out = json.loads(r.read()) assert [o["op"] for o in out["ops"]] == ["shot", "light", "mark"], out assert out["ops"][0] == {"op": "shot", "subject": "e1", "shot": "MS", "angle": "low", "focal": 35}, out assert len(out["rejected"]) == 3, out reasons = " | ".join(rj["reason"] for rj in out["rejected"]) assert "explode" in reasons and "ghost" in reasons and "disco" in reasons, reasons # prompt carries the grammar vocab + the scene roster, model + low temp set sent = llm_captured[0] prompt = " ".join(m["content"] for m in sent["messages"]) for word in ("ECU", "MWS", "dutch", "OTS-L", "85", "golden-hour", "noir", "walk-to-mark", "face-other", "two-shot-L"): assert word in prompt, f"vocab {word!r} missing from prompt" assert "e1" in prompt and "lady" in prompt and "2.5" in prompt, "roster/time missing" assert sent["model"] == "mockmodel" and sent["temperature"] <= 0.3, sent n += 1 finally: p3.terminate() p3.wait() srv3.shutdown() print(f"OK: {n} test groups passed") finally: proc.terminate() proc.wait() if __name__ == "__main__": main()