#!/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 json import os import shutil import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request 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)} 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 assert req("POST", f"/render/{rid}/end", "{}")[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 dur = subprocess.check_output( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nokey=1", str(renders / rid / "out.mp4")], text=True).strip() assert abs(float(dur) - 1.0) < 0.2, f"mp4 duration {dur}s, expected ~1s" n += 1 else: print("(render test skipped — ffmpeg/ffprobe not found)") print(f"OK: {n} test groups passed") finally: proc.terminate() proc.wait() if __name__ == "__main__": main()