#!/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 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 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 = tmp / "renders" env = {**os.environ, "SCENEGOD_ASSETS": str(assets), "SCENEGOD_SCENES": str(scenes), "SCENEGOD_RENDERS": str(renders)} 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 # --- 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 else: print("(render test skipped — ffmpeg/ffprobe not found)") # --- 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()