"""End-to-end test of the MCP over stdio with no GPU and no paid call: tests/fake_worker.py renders, tests/fake_openai.py answers the image and chat calls. Wipes and fills $REMAKE_HOME (its path must contain "test"). REMAKE_HOME=/scratch/remake_home_test python tests/e2e.py CLIP.mp4 CLIP: >= 10 s, landscape, with audio, cuts near 0.8 s and 6.1 s (the shots and keyframes below are for that clip). Flows: 1 swap, codex image mode, client-written prompts, worker max 345: one window, sheet/paint briefs, a wrong-aspect import 2 edit, API image mode: epaint, cut_windows/edit_prompt, a draft and a final, the image-call limit, quota -> STOP 3 swap, API image mode + server LLM prompts, worker max 124: a hop chain, the USD cap, a stuck hop, the too-long refusal 4 no swap worker live: windows falls back to 345, 720 is refused, wait returns at once 5 other aspects (crop): 4:3, 9:16 and 3:1 (squeezed to 2.31:1) canvases, padding, finals at the source's aspect Flows 1, 2 and 5 ingest spans that are not 17k+5 long (padded with the last shot played backwards), flow 3 one that is. """ import asyncio import json import os import re import shutil import subprocess import sys import time from contextlib import asynccontextmanager from pathlib import Path from mcp import ClientSession, StdioServerParameters, stdio_client import numpy as np from PIL import Image, ImageDraw, ImageOps HERE = Path(__file__).resolve().parent HOME = Path(os.environ["REMAKE_HOME"]).absolute() CLIP = Path(sys.argv[1]).resolve() PY = sys.executable MCP = str(Path(PY).parent / "viggle-remake-mcp") LOG = HOME / "fake_openai.jsonl" assert "test" in str(HOME), f"{HOME}: the test wipes REMAKE_HOME; use a path containing 'test'" shutil.rmtree(HOME, ignore_errors=True) (HOME / "chars").mkdir(parents=True) (HOME / "inputs").mkdir() procs = {} def start_worker(name, mx, modes="swap,edit", delay=0.0): procs[name] = subprocess.Popen([PY, HERE / "fake_worker.py", "--home", HOME, "--name", name, "--max480", str(mx), "--max736", str(mx), "--modes", modes, "--delay", str(delay)]) while not (HOME / "workers" / f"fake_{name}.json").exists(): time.sleep(0.2) def stop_worker(name): # and drop its heartbeat (a dead worker's beat stays "live" for 120 s) procs.pop(name).kill() time.sleep(0.3) (HOME / "workers" / f"fake_{name}.json").unlink() def card(path, size, text, color): im = Image.new("RGB", size, color) ImageDraw.Draw(im).text((40, 40), text, fill="white", font_size=80) im.save(path) def probe(p): """-> (frames, width, height, has audio)""" r = json.loads(subprocess.run(["ffprobe", "-v", "error", "-count_packets", "-show_entries", "stream=codec_type,nb_read_packets,width,height", "-of", "json", str(p)], capture_output=True, text=True, check=True).stdout)["streams"] v = next(s for s in r if s["codec_type"] == "video") return int(v["nb_read_packets"]), v["width"], v["height"], any(s["codec_type"] == "audio" for s in r) def frames(p): """-> uint8 (n, h, w, 3)""" _, w, h, _ = probe(p) raw = subprocess.run(["ffmpeg", "-v", "error", "-i", str(p), "-f", "rawvideo", "-pix_fmt", "rgb24", "-"], capture_output=True, check=True).stdout return np.frombuffer(raw, np.uint8).reshape(-1, h, w, 3) def requests(): return [json.loads(x) for x in LOG.read_text().splitlines()] if LOG.exists() else [] def answer(text): """Do what a CODEX IMAGE STEP says: 'generate' (a tinted copy of the last referenced picture at the brief's size) and copy it to copy_to.""" refs = json.loads(re.search(r"referenced_image_paths = (\[.*\])", text).group(1)) copy_to = Path(re.search(r"\bto (/\S+)\n", text).group(1)) W, H = map(int, json.loads(copy_to.with_suffix(".json").read_text())["size"].split("x")) im = Image.open(refs[-1]).convert("L") ImageOps.colorize(im, "#103010", "#c0f0c0").resize((W, H)).save(copy_to) return copy_to @asynccontextmanager async def server(**env): p = StdioServerParameters(command=MCP, env={"REMAKE_HOME": str(HOME), "PATH": os.environ["PATH"], **env}) async with stdio_client(p) as (r, w), ClientSession(r, w) as s: await s.initialize() async def call(tool, ok=True, **args): t0 = time.time() res = await s.call_tool(tool, args, read_timeout_seconds=900) text = "\n".join(c.text for c in res.content if c.type == "text") nimg = sum(c.type == "image" for c in res.content) assert res.is_error != ok, f"{tool}({args}) is_error={res.is_error}:\n{text[:3000]}" print(f" {time.time() - t0:5.1f}s {tool}{'' if ok else ' (expected error)'}: {text[:140]!r}{f' +{nimg} img' if nimg else ''}", flush=True) return text, nimg call.session = s yield call def plan(nf, cuts, x1=None): b = [0] + cuts + [nf] people = [{"id": "P1", "desc": "the tall man in the green elf costume and yellow tights", "replace_with": "Vela"}, {"id": "P2", "desc": "the big man in the dark red store vest", "replace_with": "none"}] if x1: people[0]["replace_with"] = "none" people.append({"id": "X1", "desc": "the white clipboard the man in the red vest holds", "replace_with": x1}) return {"people": people, "shots": [{"start": s, "end": e, "people": [p["id"] for p in people], "desc": "in a toy store"} for s, e in zip(b, b[1:])]} def where(fs, x1=False): return {str(f): {"P1": "left, facing right", "P2": "right, facing left", **({"X1": "in P2's left hand"} if x1 else {})} for f in fs} def prompt_text(n, extra=""): pics = " ".join(f" is a frame of the target video." for k in range(n)) return (f"subject_definitions:\n