#!/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, 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 # --- 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)") # --- 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") 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]] film2 = make_film("film-dissolve", {"version": 1, "shots": dshots, "audio": [{"path": "audio/test/tone.wav", "start": 0.25, "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}" 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 # --- 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()