#!/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 probe_v(f, field): """One ffprobe field off the VIDEO stream (width/height/color_primaries/...).""" return subprocess.check_output( ["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", f"stream={field}", "-of", "default=nw=1:nokey=1", str(f)], text=True).strip() def probe_rgb(f): """Top-left pixel of frame 0, decoded through the file's OWN colour tags. This is the only assertion that can catch a mislabelled encode: ffprobe will happily report color_primaries=bt709 on a file whose pixels were converted with bt601, and every player then shifts hue+contrast off what the canvas drew. Decode it back to RGB and compare with what went in.""" raw = subprocess.run(["ffmpeg", "-v", "error", "-i", str(f), "-frames:v", "1", "-f", "rawvideo", "-pix_fmt", "rgb24", "-"], capture_output=True).stdout assert len(raw) >= 3, f"could not decode a frame from {f}" return tuple(raw[0:3]) def near(got, want, tol=6): return all(abs(a - b) <= tol for a, b in zip(got, want)) 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 # a SYMLINKED bank inside the asset root — the documented way John gets his # MeshGod gallery in (`ln -s ~/MESHGOD/assets3d/gallery assets/props/...`). # Both halves used to be broken: os.walk skipped it, and safe_under's # .resolve() put it outside ASSETS so every fetch 400'd. bank = tmp / "extbank" bank.mkdir() (bank / "lamp.glb").write_bytes(b"lampbytes") os.symlink(bank, assets / "props" / "gallery") 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 # --- safe_under is lexical now: traversal still dead, symlinks live --- # (unit, in-process — the guard is the one function every path endpoint uses) sys.path.insert(0, str(ROOT)) os.environ.setdefault("SCENEGOD_ASSETS", str(assets)) from scenegod.server import safe_under, video_args, vf_args # noqa: E402 from fastapi import HTTPException # noqa: E402 for bad in ("../../etc/passwd", "a/../../x", "/etc/passwd", "..", "..\\..\\x", "a/b/../../../c"): try: safe_under(Path("/root"), bad) raise AssertionError(f"safe_under accepted {bad!r}") except HTTPException as e: assert e.status_code == 400, (bad, e.status_code) assert safe_under(Path("/root"), "characters/test/x.glb") == Path("/root/characters/test/x.glb") assert safe_under(Path("/root"), "a/./b") == Path("/root/a/b") assert safe_under(Path("/root"), "") == Path("/root") # empty relpath = the root itself # integration: the symlinked bank is listed AND streamable props = {g["path"] for g in get_json("/assets/tree")["props"]} assert "props/gallery/lamp.glb" in props, f"symlinked bank not scanned: {props}" st, b = req("GET", "/assets/file?path=props/gallery/lamp.glb") assert st == 200 and b == b"lampbytes", (st, b) # was a bare 400 "bad path" n += 1 # --- encoder options: colour-tagged, faststart, one lossy generation --- for crf, args in ((18, video_args()), (14, video_args(14))): assert args[args.index("-crf") + 1] == str(crf), args for flag, val in (("-colorspace", "bt709"), ("-color_primaries", "bt709"), ("-color_trc", "bt709"), ("-color_range", "tv")): assert args[args.index(flag) + 1] == val, args assert "+faststart" in args, args for input_only in ("-i", "-framerate", "-ss", "-t", "-f", "-y"): assert input_only not in args, f"{input_only} is an input flag: {args}" assert vf_args({"width": 320, "height": 180})[1].startswith( "scale=320:180:flags=lanczos:out_color_matrix=bt709,"), vf_args({"width": 320, "height": 180}) for m in ({"width": 320, "height": 180}, {}, {"width": 8, "height": 8}): chain = vf_args(m)[1] assert "setparams" in chain, (m, chain) # the TAG rides on every encode... # ...and so does the CONVERSION. setparams alone writes bt709 into the VUI # while swscale quietly does RGB->YUV with bt601 (see server.py CONVERT): # every mp4 mislabelled, saturated colour visibly off. Never one without # the other — this is asserted per-pixel further down, in the render tests. assert "out_color_matrix=bt709" in chain, (m, chain) assert chain.index("out_color_matrix") < chain.index("setparams"), \ f"the conversion must happen BEFORE the tag: {chain}" # no size declared = no RESIZE (the scale is there for the matrix only) assert vf_args({})[1].startswith("scale=out_color_matrix=bt709,"), vf_args({}) 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}" # colour tags: an untagged yuv420p mp4 makes QuickTime/Safari/NLEs # GUESS the primaries and shift hue+contrast off what the canvas drew assert probe_v(out, "color_primaries") == "bt709", probe_v(out, "color_primaries") assert probe_v(out, "color_transfer") == "bt709", probe_v(out, "color_transfer") # bad audio path rejected before encode assert req("POST", f"/render/{rid}/end", json.dumps({"audio": [{"path": "audio/nope.wav"}]}))[0] == 400 n += 1 # --- supersample: frames arrive 2x, ffmpeg delivers the declared size --- big = tmp / "bigframes" big.mkdir() ff("-f", "lavfi", "-i", "testsrc=size=640x360:rate=30", "-frames:v", "10", str(big / "%06d.png")) r3 = json.loads(req("POST", "/render/begin", json.dumps( {"name": "ss", "fps": 30, "width": 320, "height": 180}))[1])["renderId"] for i, png in enumerate(sorted(big.glob("*.png"))): assert req("POST", f"/render/{r3}/frame/{i}", png.read_bytes())[0] == 200 assert req("POST", f"/render/{r3}/end", json.dumps({}))[0] == 200 for _ in range(150): s3 = json.loads(req("GET", f"/render/{r3}/status")[1])["state"] if s3 in ("done", "error"): break time.sleep(0.2) assert s3 == "done", (s3, json.loads(req("GET", f"/render/{r3}/status")[1])["log"][-600:]) ssout = renders / r3 / "out.mp4" assert probe_v(ssout, "width") == "320" and probe_v(ssout, "height") == "180", \ (probe_v(ssout, "width"), probe_v(ssout, "height")) assert probe_v(ssout, "color_primaries") == "bt709" # x264 stamps its settings in an SEI — a deliverable is crf 18 assert b"crf=18.0" in ssout.read_bytes(), "deliverable should encode at crf 18" # and -vf coexists with the audio -filter_complex (different streams) r4 = json.loads(req("POST", "/render/begin", json.dumps( {"name": "ssa", "fps": 30, "width": 320, "height": 180}))[1])["renderId"] for i, png in enumerate(sorted(big.glob("*.png"))): assert req("POST", f"/render/{r4}/frame/{i}", png.read_bytes())[0] == 200 assert req("POST", f"/render/{r4}/end", json.dumps( {"audio": [{"path": "audio/test/tone.wav", "start": 0, "gain": 1.0}], "intermediate": True}))[0] == 200 for _ in range(150): s4 = json.loads(req("GET", f"/render/{r4}/status")[1])["state"] if s4 in ("done", "error"): break time.sleep(0.2) assert s4 == "done", (s4, json.loads(req("GET", f"/render/{r4}/status")[1])["log"][-600:]) sa = renders / r4 / "out.mp4" assert probe_v(sa, "width") == "320", probe_v(sa, "width") assert seg_db(sa, 0.05, 0.2) > -35, "audio lost when -vf and -filter_complex coexist" # film SHOTS are intermediates: finer crf so the concat's re-encode is # the only generation that meaningfully quantises (film.js sets this) assert b"crf=14.0" in sa.read_bytes(), "intermediate shot should encode at crf 14" n += 1 # --- COLOUR: the pixels must agree with the tags ---------------- # The tags-only version of this test passed while every render was # converted RGB->YUV with bt601 under a bt709 label: pure green stored # Y=145 (bt709 green is 173) and decoded back as (0,216,0). Assert the # ROUND TRIP, per pixel, on both filter paths — that is the only thing # that can tell a correct encode from a confidently mislabelled one. def render_pngs(begin_body, src, count=5, end_body="{}"): r = json.loads(req("POST", "/render/begin", json.dumps(begin_body))[1])["renderId"] for i in range(count): assert req("POST", f"/render/{r}/frame/{i}", Path(src).read_bytes())[0] == 200 assert req("POST", f"/render/{r}/end", end_body)[0] == 200 for _ in range(150): stt = json.loads(req("GET", f"/render/{r}/status")[1]) if stt["state"] in ("done", "error"): break time.sleep(0.2) return r, stt for name, want in (("0x00FF00", (0, 255, 0)), # the loudest case: bt601 green is 24 IRE low ("0xE62878", (230, 40, 120)), # neon magenta — this app's actual material ("0x3C78DC", (60, 120, 220))): # sky blue flat = tmp / f"flat{name}.png" ff("-f", "lavfi", "-i", f"color=c={name}:s=256x144", "-frames:v", "1", str(flat)) # sized path (the supersample downscale: 256x144 frames -> 128x72) rc, stt = render_pngs({"name": "col", "fps": 30, "width": 128, "height": 72}, flat) assert stt["state"] == "done", stt got = probe_rgb(renders / rc / "out.mp4") assert near(got, want), (f"{name} scaled render decodes as {got}, expected ~{want} " "— pixels converted with the wrong matrix for the tag") # no-size path (begin body without width/height: setparams only + iw:ih scale) rd, stt = render_pngs({"name": "col2", "fps": 30}, flat) assert stt["state"] == "done", stt got = probe_rgb(renders / rd / "out.mp4") assert near(got, want), f"{name} unscaled render decodes as {got}, expected ~{want}" # ...and that path's scale really is matrix-only: no resize to any default assert probe_v(renders / rd / "out.mp4", "width") == "256", \ "a begin body without width/height must not resize the frames" n += 1 # --- a capture SMALLER than the declared size is an ERROR ------- # (it used to be silently UPSCALED into a soft mp4 with state=done) small = tmp / "small.png" ff("-f", "lavfi", "-i", "color=c=red:s=64x36", "-frames:v", "1", str(small)) rs, stt = render_pngs({"name": "undersized", "fps": 30, "width": 320, "height": 180}, small) assert stt["state"] == "error", f"undersized frames encoded anyway: {stt}" assert "64x36" in stt["log"] and "320x180" in stt["log"], stt["log"] assert not (renders / rs / "out.mp4").exists(), "no mp4 should be written" # bigger is normal (that IS the supersample) and equal is fine for w, h in ((128, 72), (256, 144)): _, stt = render_pngs({"name": "okay", "fps": 30, "width": w, "height": h}, tmp / "flat0x00FF00.png") assert stt["state"] == "done", (w, h, stt) 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") # the film is the file anyone actually watches — it must be tagged too assert probe_v(out, "color_primaries") == "bt709", probe_v(out, "color_primaries") assert probe_v(out, "color_transfer") == "bt709", probe_v(out, "color_transfer") # ...and TAGGED IS NOT CONVERTED. A film assembled from shots that are # not themselves bt709 (an older build's mp4 still sitting in renders/, # anything hand-dropped) used to inherit their bt601 pixels under the # film's bt709 tag: green came out (0,216,0). These fixtures are exactly # that case — untagged lavfi mp4s — so this discriminates. filmc = make_film("film-colour", {"version": 1, "shots": shots}) outc = finish_film(filmc["filmId"], [fake_shot_render(2.0, color="0x00FF00"), fake_shot_render(3.0, color="0x00FF00")]) gotc = probe_rgb(outc) assert near(gotc, (0, 255, 0)), f"film decodes green as {gotc}, expected ~(0,255,0)" 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()