586a3ecbd2
templates/ ships with the repo (two-hander-golden-hour, record-store,
empty-stage), GET /templates + /templates/{name} behind the same slug and
path-traversal guards as /scenes; presentation metadata rides in a template
key that validate_scene ignores and toJSON drops, so loading then saving
yields plain scene JSON.
Test group asserts the three list in order, round-trip, traversal/unknown
rejected, write verbs refused, and every template passes validate_scene with
every referenced asset present on disk - the check that stops a rotten
template shipping. 10 groups green.
Verified live: two-hander loads 7 entities, 8s, both performers animating,
cut fires cam1 to cam2 at 4.5s; other two load clean.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
361 lines
18 KiB
Python
361 lines
18 KiB
Python
#!/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
|
|
|
|
# --- 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
|
|
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()
|