monster/SCENEGOD
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
ba29d0c6d6
SCENEGOD/scripts/test_server.py
type-two fb59601cc3 [laneC] /rhubarb viseme endpoint (ffmpeg-style 503 gating) 
GET /rhubarb?path= -> rhubarb -f json -> mouthCues for Lane A's viseme lane;
503 when the binary is absent (not installed on ultra yet). Path-guarded.
test_server.py: +503 gating assert (9 groups green).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-18 21:10:38 +10:00

247 lines
11 KiB
Python
Raw Blame History

#!/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)}
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()
print(f"OK: {n} test groups passed")
finally:
proc.terminate()
proc.wait()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink