#!/usr/bin/env python3 """Self-test for higgs_vclaw_adapter — verifies the videoclaw-v3 I/O contract WITHOUT touching CloakBrowser or rendering anything live. It does two things: 1. In-process: drives dispatch() with a STUB HiggsSession (no browser, no network) and asserts the SUBMIT/POLL/CANCEL stdout shapes match what videoclaw-v3's execution-runtime parses (externalJobId+rawResult on submit; status/outputs[].{id,kind,path,sceneIndex,backend}/issues on poll; status='cancelled' on cancel). Also verifies job-state round-trips: submit writes /.vclaw-jobs/.json and poll reads it. 2. Subprocess: pipes a fake SUBMIT payload through the REAL wrapper script with the session factory monkeypatched off (via HIGGS_VCLAW_TEST_STUB=1) to prove the stdin->stdout JSON transport works end-to-end as videoclaw drives it. Run: ./.venv/bin/python test_adapter.py (plain `python3 test_adapter.py` also works — no third-party deps.) """ import contextlib import json import os import subprocess import sys import tempfile HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import adapter # noqa: E402 # --------------------------------------------------------------------------- # Stub session — replaces CloakBrowser/Higgsfield with deterministic responses. # --------------------------------------------------------------------------- class StubSession: """Records submits, fakes a wallet that never drops, and serves completed jobs with a fake result URL so poll downloads a placeholder file.""" class _Page: """No-op page. Real submits pace themselves with wait_for_timeout to stay under Cloudflare Turnstile; the stub records the pacing without spending SUBMIT_SPACING_S x N seconds of test time.""" def __init__(self): self.waits = [] def wait_for_timeout(self, ms): self.waits.append(ms) def reload(self, **kw): pass def __init__(self, poll_status="completed"): self.page = self._Page() self.submitted = [] self.poll_status = poll_status self.cancelled = [] self.downloads = [] self.uploads = [] self.closed = False def wallet(self): # credits_balance MUST stay flat (free path); subscription can move. return {"credits_balance": 100, "subscription_balance": 50} def upload_image(self, path, mime="image/png"): # Record what would be uploaded (start frame) and hand back a fake media. self.uploads.append({"path": path, "mime": mime, "existed": os.path.exists(path)}) return ({"id": f"media-{len(self.uploads)}", "url": f"https://example.invalid/media-{len(self.uploads)}.png"}, None) def upload_video(self, path, mime="video/mp4"): self.uploads.append({"path": path, "mime": mime, "existed": os.path.exists(path)}) return ({"id": f"video-{len(self.uploads)}", "url": f"https://example.invalid/video-{len(self.uploads)}.mp4", "type": "video_input", "width": 720, "height": 1280, "duration": 8.0}, None) def upload_audio(self, path, mime="audio/mpeg"): self.uploads.append({"path": path, "mime": mime, "existed": os.path.exists(path)}) return ({"id": f"audio-{len(self.uploads)}", "url": f"https://example.invalid/audio-{len(self.uploads)}.mp3", "type": "audio_input", "name": "v", "extension": "mp3"}, None) def submit_scene(self, params): jsid = f"jobset-{len(self.submitted)}" self.submitted.append({"jsid": jsid, "params": params}) return jsid, None def query_job(self, job_set_id): if self.poll_status == "completed": job = { "id": f"job-{job_set_id}", "status": "completed", "results": {"raw": {"url": f"https://example.invalid/{job_set_id}.mp4"}}, } elif self.poll_status == "pending": job = {"id": f"job-{job_set_id}", "status": "queued", "results": {}} else: # failed job = {"id": f"job-{job_set_id}", "status": "nsfw", "results": {}} return job, {"jobs": [job]} def cancel_job(self, job_set_id): self.cancelled.append(job_set_id) return {"status": 200, "body": {"ok": True}} def download(self, url, path): os.makedirs(os.path.dirname(path) or ".", exist_ok=True) with open(path, "wb") as f: f.write(b"FAKEMP4") self.downloads.append((url, path)) return 7 def close(self): self.closed = True def _payload(output_dir, n_scenes=2, refs=None): return { "workspaceRoot": output_dir, "projectSlug": "selftest", "routeId": "seedance-direct", "outputDir": output_dir, "executionProfile": { "aspectRatio": "16:9", "quality": "fast", "resolution": "720p", "generateAudio": True, "outputCount": 1, }, "tasks": [ { "sceneIndex": i, "prompt": f"scene {i} prompt", "durationSeconds": 8, "referencePaths": (refs or []) if i == 0 else [], } for i in range(n_scenes) ], } def _check(cond, msg): if not cond: raise AssertionError(msg) @contextlib.contextmanager def allow_t2v_fallback(): """Opt back into the pre-guard behaviour for tests that exercise DEGRADATION. Submitting without a requested start frame is now a hard refusal, because a dropped identity anchor produces a clip that passes every automated check while showing the wrong subject. The degrade path still exists behind this env var, and both branches are covered — see test_missing_start_frame_refuses_by_default. """ prev = os.environ.get("HIGGS_VCLAW_ALLOW_T2V_FALLBACK") os.environ["HIGGS_VCLAW_ALLOW_T2V_FALLBACK"] = "1" try: yield finally: if prev is None: os.environ.pop("HIGGS_VCLAW_ALLOW_T2V_FALLBACK", None) else: os.environ["HIGGS_VCLAW_ALLOW_T2V_FALLBACK"] = prev # --------------------------------------------------------------------------- # Test 1: in-process SUBMIT -> POLL round-trip with a stub session. # --------------------------------------------------------------------------- def test_submit_poll_roundtrip(): with tempfile.TemporaryDirectory() as d: stub = StubSession(poll_status="completed") with allow_t2v_fallback(): # bogus ref path: this test is about the round-trip sub = adapter.action_submit(_payload(d, n_scenes=2, refs=["/x/ref.png"]), session_factory=lambda: stub) # --- SUBMIT stdout shape (execution-runtime requires externalJobId) --- _check(isinstance(sub.get("externalJobId"), str) and sub["externalJobId"], "submit must return a non-empty string externalJobId") raw = sub["rawResult"] _check(raw["externalJobId"] == sub["externalJobId"], "rawResult.externalJobId must equal top-level externalJobId") _check(len(raw["submittedScenes"]) == 2, "two submittedScenes expected") for ss in raw["submittedScenes"]: _check("sceneIndex" in ss and "taskId" in ss, "each submittedScene needs sceneIndex + taskId") _check(len(stub.submitted) == 2, "stub should have received two submits") # taskId must be the Higgsfield job_set id we returned. _check(raw["submittedScenes"][0]["taskId"] == "jobset-0", "taskId must be the job_set id") # Free-safety: no spend issue because wallet stayed flat. _check(not any("FREE-SAFETY" in i for i in raw["issues"]), "no free-safety abort expected when credits stay flat") # Phase-1 ref limitation surfaced for scene 0. _check(any("referencePath" in i for i in raw["issues"]), "reference-ignored issue expected for scene with refs") # --- job state written where poll will read it --- js_path = adapter.job_state_path(d, sub["externalJobId"]) _check(os.path.exists(js_path), "submit must write job-state file") state = json.load(open(js_path)) _check(state["routeId"] == "seedance-direct", "routeId pinned in state") _check(len(state["scenes"]) == 2, "two scenes persisted") _check(state["scenes"][0]["taskId"] == "jobset-0", "scene taskId persisted") # --- POLL stdout shape (VideoExecutionPollResult) --- stub2 = StubSession(poll_status="completed") poll = adapter.action_poll( {"action": "poll", "outputDir": d, "externalJobId": sub["externalJobId"], "workspaceRoot": d}, session_factory=lambda: stub2, ) _check(poll["status"] == "completed", f"expected completed, got {poll['status']}") _check(poll["externalJobId"] == sub["externalJobId"], "poll echoes externalJobId") _check(len(poll["outputs"]) == 2, "two outputs expected") for o in poll["outputs"]: # These are the exact fields execution-runtime filters on. _check(isinstance(o["id"], str) and o["id"], "output.id non-empty string") _check(o["kind"] == "video", "output.kind must be 'video'") _check(isinstance(o["path"], str) and os.path.exists(o["path"]), "output.path must be a downloaded file") _check(isinstance(o["sceneIndex"], int), "output.sceneIndex int") _check(o["backend"] == "seedance-direct", "output.backend pinned") _check(isinstance(poll["issues"], list), "issues must be a list") # Downloaded to outputDir/scene-.mp4 exactly. _check(os.path.exists(os.path.join(d, "scene-0.mp4")), "scene-0.mp4 downloaded") _check(os.path.exists(os.path.join(d, "scene-1.mp4")), "scene-1.mp4 downloaded") # --- POLL idempotency: existing file not re-downloaded --- stub3 = StubSession(poll_status="completed") adapter.action_poll( {"action": "poll", "outputDir": d, "externalJobId": sub["externalJobId"], "workspaceRoot": d}, session_factory=lambda: stub3, ) _check(len(stub3.downloads) == 0, "second poll must NOT re-download already-present scenes") print("PASS test_submit_poll_roundtrip") # --------------------------------------------------------------------------- # Test 2: POLL pending + failed statuses map correctly. # --------------------------------------------------------------------------- def test_poll_pending_and_failed(): with tempfile.TemporaryDirectory() as d: sub = adapter.action_submit(_payload(d, n_scenes=1), session_factory=lambda: StubSession()) eid = sub["externalJobId"] pending = adapter.action_poll( {"action": "poll", "outputDir": d, "externalJobId": eid, "workspaceRoot": d}, session_factory=lambda: StubSession(poll_status="pending")) _check(pending["status"] == "pending", "pending status expected") _check(pending["outputs"] == [], "no outputs while pending") failed = adapter.action_poll( {"action": "poll", "outputDir": d, "externalJobId": eid, "workspaceRoot": d}, session_factory=lambda: StubSession(poll_status="failed")) _check(failed["status"] == "failed", "failed status expected (nsfw)") _check(any("nsfw" in i.lower() for i in failed["issues"]), "failure reason surfaced in issues") print("PASS test_poll_pending_and_failed") # --------------------------------------------------------------------------- # Test 3: CANCEL -> 'unsupported' (Higgsfield has no cancel; no browser opened). # --------------------------------------------------------------------------- def test_cancel(): with tempfile.TemporaryDirectory() as d: sub = adapter.action_submit(_payload(d, n_scenes=2), session_factory=lambda: StubSession()) eid = sub["externalJobId"] def _boom(): raise AssertionError("cancel must NOT open a session (no cancel endpoint)") res = adapter.action_cancel( {"action": "cancel", "outputDir": d, "externalJobId": eid, "workspaceRoot": d}, session_factory=_boom) _check(res["status"] == "unsupported", "cancel status must be 'unsupported' (Higgsfield has no cancel)") _check(res["externalJobId"] == eid, "cancel echoes externalJobId") _check(isinstance(res["issues"], list) and res["issues"], "cancel explains why it is unsupported in issues") # State must be UNTOUCHED — the job keeps running and stays pollable. state = json.load(open(adapter.job_state_path(d, eid))) _check(all(s["status"] == "submitted" for s in state["scenes"]), "cancel must not falsify scene state to failed") print("PASS test_cancel") def test_wallet_is_read_only_and_closes_session(): class WalletSession(StubSession): def __init__(self): super().__init__() self.closed = False def close(self): self.closed = True stub = WalletSession() result = adapter.dispatch({"action": "wallet"}, session_factory=lambda: stub) _check(result["credits_balance"] == 100, "wallet action must return credits_balance") _check(result["subscription_balance"] == 50, "wallet action must return subscription_balance") _check(stub.submitted == [], "wallet action must never submit a scene") _check(stub.closed, "wallet action must close the authenticated session") print("PASS test_wallet_is_read_only_and_closes_session") # --------------------------------------------------------------------------- # Test 4: FREE-SAFETY — a wallet whose credits drop must HARD-ABORT (raise). # --------------------------------------------------------------------------- def test_free_safety_abort(): class SpendingSession(StubSession): def __init__(self): super().__init__() self._calls = 0 def wallet(self): # before=100, after=90 -> a paid spend must hard-abort. self._calls += 1 return {"credits_balance": 100 if self._calls == 1 else 90, "subscription_balance": 50} with tempfile.TemporaryDirectory() as d: raised = None try: adapter.action_submit(_payload(d, n_scenes=1), session_factory=lambda: SpendingSession()) except RuntimeError as e: raised = e _check(raised is not None and "FREE-SAFETY" in str(raised), "credits drop must raise a FREE-SAFETY RuntimeError") # Forensic job-state must still have been persisted before the abort. jobs_dir = adapter.job_state_dir(d) _check(os.path.isdir(jobs_dir) and os.listdir(jobs_dir), "free-safety abort must still persist forensic job-state") print("PASS test_free_safety_abort") class _FakeVideoSession: """Drives the REAL HiggsSession.upload_video with a scripted transport. Stubbing upload_video itself would test nothing — the whole point is the parsing, and specifically the dict-vs-list branch that would crash the image path. So this fakes only api()/ctx/page and lets the real method run. """ class _Req: def __init__(self, status=200): self.status = status self.puts = [] def put(self, url, data=None, headers=None): self.puts.append({"url": url, "bytes": len(data or b""), "headers": headers}) return type("R", (), {"status": self.status})() class _Page: def wait_for_timeout(self, ms): pass def __init__(self, create_body, create_status=200, finalize_body=None, finalize_status=200, poll_body=None, put_status=200): self.calls = [] self._create = (create_status, create_body) self._finalize = (finalize_status, finalize_body if finalize_body is not None else {"width": 720, "height": 1280, "duration": 8.0}) self._poll = poll_body if poll_body is not None else { "status": "uploaded", "ip_check_finished": True, "type": "video_input"} self.ctx = type("C", (), {"request": self._Req(put_status)})() self.page = self._Page() def api(self, method, path, body=None): self.calls.append({"method": method, "path": path, "body": body}) if method == "POST" and path == adapter.VIDEO_PATH: return {"status": self._create[0], "body": self._create[1]} if method == "POST" and path.endswith("/upload"): return {"status": self._finalize[0], "body": self._finalize[1]} if method == "GET": return {"status": 200, "body": self._poll} return {"status": 404, "body": None} CAPTURED_CREATE = { # verbatim from the 2026-08-07 capture "id": "6e73ff05-c59a-40ba-b01a-51276a90f213", "url": "https://d2ol7oe51mr4n9.cloudfront.net/user_x/6e73ff05.mp4", "upload_url": "https://d276s3zg8h21b2.cloudfront.net/user_x/6e73ff05.mp4?X-Amz-Signature=x", } def _video_file(d, name="w08.mp4"): p = os.path.join(d, name) with open(p, "wb") as f: f.write(b"\x00\x00\x00 ftypisom" + b"\x00" * 512) return p def test_video_upload_parses_the_dict_response(): """The captured create returns a DICT where /media/batch returns a LIST. upload_image does `(body or [None])[0]`, which raises on a dict. This is the exact shape higgs-cloak's ARCHITECTURE.md suspected and never confirmed. """ with tempfile.TemporaryDirectory() as d: sess = _FakeVideoSession(CAPTURED_CREATE) media, reason = adapter.HiggsSession.upload_video(sess, _video_file(d)) _check(reason is None, f"captured shape must parse; got {reason}") _check(media["id"] == CAPTURED_CREATE["id"], "must return the media id") _check(media["type"] == "video_input", f"must report type video_input (not media_input); got {media.get('type')}") _check(media["width"] == 720 and media["height"] == 1280, "must carry dimensions") # The request bodies must match the capture exactly — this is the contract. create = [c for c in sess.calls if c["path"] == adapter.VIDEO_PATH][0] _check(create["body"] == {"mimetype": "video/mp4", "force_ip_check": True}, f"create body must match capture; got {create['body']}") fin = [c for c in sess.calls if c["path"].endswith("/upload")][0] _check(fin["body"]["surface"] == "seedance_2", f"surface must be seedance_2 (same as images); got {fin['body'].get('surface')}") _check(fin["body"]["filename"] == "w08.mp4", "must send the basename") # The presigned URL signs content-type;host — a wrong header breaks the sig. put = sess.ctx.request.puts[0] _check(put["headers"]["content-type"] == "video/mp4", f"PUT must send content-type video/mp4; got {put['headers']}") print("PASS test_video_upload_parses_the_dict_response") def test_video_upload_tolerates_a_list_response(): """Defensive: if the endpoint ever matches the image shape, do not break.""" with tempfile.TemporaryDirectory() as d: sess = _FakeVideoSession([CAPTURED_CREATE]) media, reason = adapter.HiggsSession.upload_video(sess, _video_file(d)) _check(reason is None and media["id"] == CAPTURED_CREATE["id"], f"single-item list must also parse; got {reason}") print("PASS test_video_upload_tolerates_a_list_response") def test_upload_waits_for_the_ip_check_not_just_the_bytes(): """`status: "uploaded"` means the bytes landed, NOT that the media is usable. The finalize response literally returns {"status": "uploaded", "ip_check_finished": false}. Treating that as ready submits early and the job is rejected with HTTP 400 {"error_type":"other","text":"IP check not finished for input media"}. upload_image was always correct here; upload_video briefly was not. """ with tempfile.TemporaryDirectory() as d: # Poll keeps saying "uploaded" with the check unfinished -> must NOT return. sess = _FakeVideoSession( CAPTURED_CREATE, poll_body={"status": "uploaded", "ip_check_finished": False}) media, reason = adapter.HiggsSession.upload_video(sess, _video_file(d)) _check(media is None, "a media whose IP check has not finished must not be returned as usable") _check(reason and "not ready" in reason, f"must say it timed out waiting; got {reason}") # Once the check finishes, it is usable. ok = _FakeVideoSession( CAPTURED_CREATE, poll_body={"status": "uploaded", "ip_check_finished": True, "type": "video_input"}) media2, reason2 = adapter.HiggsSession.upload_video(ok, _video_file(d)) _check(media2 is not None and reason2 is None, f"a finished check must yield the media; got {reason2}") print("PASS test_upload_waits_for_the_ip_check_not_just_the_bytes") def test_video_under_640_is_refused_with_the_reason(): """A small clip uploads fine and then fails opaquely at submit. Higgsfield needs >=640 on both edges. Catching it here turns a confusing downstream rejection into a sentence naming the actual dimensions. """ with tempfile.TemporaryDirectory() as d: sess = _FakeVideoSession(CAPTURED_CREATE, finalize_body={"width": 512, "height": 512, "duration": 8.0}) media, reason = adapter.HiggsSession.upload_video(sess, _video_file(d)) _check(media is None, "a sub-640 video must not be returned as usable") _check("512x512" in reason and "640" in reason, f"reason must name the real and required sizes; got {reason}") print("PASS test_video_under_640_is_refused_with_the_reason") def test_video_path_404_explains_the_prefix_mapping(): """The one inferred bit of the flow must fail legibly. The capture ran on the gateway host where the path is /fnf/video; this adapter talks to a host that drops the /fnf prefix. If that mapping is wrong, say so rather than emitting a bare 404 someone has to re-derive. """ with tempfile.TemporaryDirectory() as d: sess = _FakeVideoSession(None, create_status=404) media, reason = adapter.HiggsSession.upload_video(sess, _video_file(d)) _check(media is None, "a 404 must not yield a media") _check("/fnf" in reason and "HIGGS_VCLAW_VIDEO_PATH" in reason, f"404 must explain the prefix mapping and the override; got {reason}") print("PASS test_video_path_404_explains_the_prefix_mapping") def test_video_rejected_by_moderation_surfaces_status(): with tempfile.TemporaryDirectory() as d: sess = _FakeVideoSession(CAPTURED_CREATE, poll_body={"status": "nsfw", "ip_check_finished": True}) media, reason = adapter.HiggsSession.upload_video(sess, _video_file(d)) _check(media is None and "nsfw" in reason, f"moderation rejection must surface the status; got {reason}") print("PASS test_video_rejected_by_moderation_surfaces_status") def test_api_url_targets_the_datadome_allowlisted_host(): """The old host was not on DataDome's fetch-interception allowlist. window.ddoptions.ajaxListenerPath is [{"host": "fnf-api-gw.higgsfield.ai"}] — one host, and the engine's previous base (fnf.higgsfield.ai) was not it, so our requests never carried the session header the UI sends on every call. """ _check("fnf-api-gw.higgsfield.ai" in adapter.API_BASE, f"base must be the allowlisted host; got {adapter.API_BASE!r}") _check(adapter.API_PREFIX == "/fnf", f"gateway paths are prefixed /fnf; got {adapter.API_PREFIX!r}") url = adapter.API_BASE + adapter.API_PREFIX + "/jobs/v2/seedance_2_5" _check(url == "https://fnf-api-gw.higgsfield.ai/fnf/jobs/v2/seedance_2_5", f"submit url must match the captured UI submit; got {url}") # The prefix must be applied exactly once — a doubled /fnf/fnf 404s. _check(url.count("/fnf/") == 1, f"prefix applied twice: {url}") # And it must reach the fetch, not just sit in a constant. _check(adapter.API_BASE + adapter.API_PREFIX in adapter._FETCH_JS, "the in-page fetch must use base+prefix") print("PASS test_api_url_targets_the_datadome_allowlisted_host") def test_fetch_surfaces_bot_protection_response_headers(): """Response headers used to be discarded, which is why DataDome's rotation was invisible while we theorised about rate limits for hours.""" js = adapter._FETCH_JS _check("ddHeaders" in js, "the fetch must return ddHeaders") _check("x-(datadome|set-cookie|dd-b)" in js, "must capture the datadome/x-set-cookie/x-dd-b headers") _check("chars>" in js, "x-set-cookie is a session token — record its LENGTH, never its value") print("PASS test_fetch_surfaces_bot_protection_response_headers") def test_submits_are_paced_to_stay_under_turnstile(): """Three back-to-back submits tripped Cloudflare Turnstile. Measured 2026-08-07 on a 10-scene batch: EXACTLY the first 3 were accepted and the remaining 7 came back 403 turnstile_required. Same live session, same model — anti-bot rate limiting, not a dead session. Pacing prevents it; the retry in submit_scene recovers when it fires anyway. """ with tempfile.TemporaryDirectory() as d: stub = StubSession() adapter.action_submit(_payload(d, n_scenes=4), session_factory=lambda: stub) _check(len(stub.submitted) == 4, "all four scenes should submit") # 3 gaps for 4 scenes — the first must NOT wait, so a single-scene submit # pays nothing for this. pacing = [w for w in stub.page.waits if w >= adapter.SUBMIT_SPACING_S * 1000] _check(len(pacing) == 3, f"expected 3 inter-scene waits for 4 scenes; got {len(pacing)}") solo = StubSession() adapter.action_submit(_payload(d, n_scenes=1), session_factory=lambda: solo) solo_pacing = [w for w in solo.page.waits if w >= adapter.SUBMIT_SPACING_S * 1000] _check(not solo_pacing, f"a single-scene submit must not pace at all; got {solo_pacing}") print("PASS test_submits_are_paced_to_stay_under_turnstile") def test_duration_clamps_to_the_model_window(): """Production Seedance 2.5 uses the current 20s ceiling; 2.0 stays at 15s. There was NO clamp — a duration outside the model's window is a 422 on EVERY submit, which is exactly how an overnight batch died on its first night via the resolution field. Same class of bug, same treatment. """ _check(adapter.clamp_duration(20, "seedance_2_5") == 20, "2.5 must allow the production 20s ceiling") _check(adapter.clamp_duration(45, "seedance_2_5") == 20, "over-long must clamp to 20, not 422") _check(adapter.clamp_duration(30, "seedance_2_0") == 15, "the 2.0 family is capped at 15s") _check(adapter.clamp_duration(1, "seedance_2_5") == 4, "under the floor must clamp UP to 4s") # An unknown model takes the conservative window: a short clip is a visible, # recoverable disappointment; a 422 is a dead batch. _check(adapter.clamp_duration(30, "some_future_model") == 15, "an unknown model must get the conservative window") _check(adapter.clamp_duration(None, "seedance_2_5") == 8, "None falls back to 8s") _check(adapter.clamp_duration("nonsense", "seedance_2_5") == 8, "unparseable duration must not raise") p = adapter.build_params({"aspectRatio": "9:16"}, "hi", 20) _check(p["duration"] == 20, f"a 20s request must survive to params; got {p['duration']}") print("PASS test_duration_clamps_to_the_model_window") def test_submit_body_sends_the_mode_not_the_model_name(): """params.model is a MODE; the MODEL name selects the endpoint. Sending the model name in the body is HTTP 422, and the server says so: {"loc":["model"],"msg":"Input should be 'default', 'video_edit' or 'video_extension'","input":"seedance_2_5"} That cost a live submit to discover even though the captured UI payload had "model":"default" in plain sight. Pinned so it cannot regress. """ params = adapter.build_params({"aspectRatio": "9:16"}, "hi", 8) _check(params["model"] in adapter.FREE_MODELS, "build_params keeps the real model, which submit_scene uses for the URL") body = adapter.submit_body(params) _check(body["params"]["model"] == "default", f"body must carry the MODE; got {body['params']['model']!r}") _check(body["params"]["use_unlim"] is True and body["use_unlim"] is True, "use_unlim must ride both inside params and at the top, as the UI sends it") _check(body["params"]["batch_size"] == 1, "captured submits carry batch_size 1") _check(body["use_free_gens"] is False, "free-gens must stay off") print("PASS test_submit_body_sends_the_mode_not_the_model_name") def test_voice_goes_on_the_audio_channel_not_as_a_black_frame_video(): """The voice must ride role:"audio", not role:"video". The black-frame-video hack is inherited from a lane that had no audio field. Higgsfield has one, and its moderation REJECTS some of those black-frame videos outright — one window was refused six times as a video media and accepted first time as audio. The prompt was never the problem. """ with tempfile.TemporaryDirectory() as d: img = os.path.join(d, "portrait.jpg") open(img, "wb").write(b"\xff\xd8\xff" + b"\x00" * 64) # A bare .mp3 exercises the audio path without needing ffmpeg — and a # voice legitimately arrives in that form, which is why split_references # now recognises audio extensions as voices. voice = os.path.join(d, "w13.mp3") open(voice, "wb").write(b"ID3" + b"\x00" * 256) payload = _payload(d, n_scenes=1) payload["tasks"][0]["referencePaths"] = [img, voice] payload["tasks"][0]["referenceRole"] = "character" stub = StubSession() with allow_t2v_fallback(): # the seed is a stub path; this test is about ROLE adapter.action_submit(payload, session_factory=lambda: stub) params = stub.submitted[0]["params"] roles = [m["role"] for m in params["medias"]] _check("audio" in roles, f"the voice must be attached on the audio channel; roles={roles}") _check(roles.count("audio") == 1, f"the voice must be attached exactly once; roles={roles}") _check("video" not in roles, f"no black-frame video media should be sent; roles={roles}") a = [m for m in params["medias"] if m["role"] == "audio"][0] _check(a["data"].get("type") == "audio_input", f"audio media must be typed audio_input; got {a['data'].get('type')}") _check("<<>>" in params["prompt"], f"the prompt must cite the audio media; got {params['prompt'][:90]!r}") _check("<<>>" not in params["prompt"], f"one voice must not create a second citation; got {params['prompt']!r}") print("PASS test_voice_goes_on_the_audio_channel_not_as_a_black_frame_video") def test_media_entries_route_by_media_type(): """build_media_entries must key the role off the media TYPE, not position.""" aud = {"id": "a", "type": "audio_input"} vid = {"id": "v", "type": "video_input"} medias, tokens = adapter.build_media_entries(videos=[aud, vid]) _check(medias[0]["role"] == "audio" and medias[1]["role"] == "video", f"roles must follow the media type; got {[m['role'] for m in medias]}") _check(tokens == ["<<>>", "<<>>"], f"each role numbers from 1 independently; got {tokens}") print("PASS test_media_entries_route_by_media_type") def test_voice_clip_rides_alongside_the_start_frame(): """A rap-MV task is [portrait.jpg, voice.mp4] — BOTH must survive. The old path picked exactly one reference and tail-framed any video into a still, so it took the portrait as the i2v start frame and silently discarded the voice. For a black-frame voice clip the tail frame is a black rectangle, so had the order been reversed it would have opened on black instead. """ with tempfile.TemporaryDirectory() as d: img = os.path.join(d, "portrait.jpg") with open(img, "wb") as f: f.write(b"\xff\xd8\xff" + b"\x00" * 64) voice = os.path.join(d, "w08.mp4") with open(voice, "wb") as f: f.write(b"\x00\x00\x00 ftypisom" + b"\x00" * 256) payload = _payload(d, n_scenes=1) payload["tasks"][0]["referencePaths"] = [img, voice] payload["tasks"][0]["referenceRole"] = "character" payload["tasks"][0]["prompt"] = "the performer on a neon rooftop" stub = StubSession() with allow_t2v_fallback(): # the seed is a stub path; this test is about ROLE adapter.action_submit(payload, session_factory=lambda: stub) params = stub.submitted[0]["params"] roles = [m["role"] for m in params["medias"]] _check("video" in roles, f"the voice clip must be attached as a video media; roles={roles}") vid = [m for m in params["medias"] if m["role"] == "video"][0] _check(vid["data"].get("type") == "video_input", f"video media must carry type video_input; got {vid['data'].get('type')}") _check("<<>>" in params["prompt"], f"the prompt must cite the voice media; got {params['prompt']!r}") _check("neon rooftop" in params["prompt"], "the operator prompt must survive") print("PASS test_voice_clip_rides_alongside_the_start_frame") def test_chain_relay_seed_is_not_treated_as_voice(): """THE regression guard for auto-chain. A relay seed is also a bare .mp4. If it were attached as a video media the chain would break: the next scene would stop seeding from the previous scene's last frame. referenceRole 'keyframe' must keep the tail-frame path. """ with tempfile.TemporaryDirectory() as d: prev = os.path.join(d, "scene-0.mp4") with open(prev, "wb") as f: f.write(b"\x00\x00\x00 ftypisom" + b"\x00" * 256) payload = _payload(d, n_scenes=1) payload["tasks"][0]["referencePaths"] = [prev] payload["tasks"][0]["referenceRole"] = "keyframe" stub = StubSession() with allow_t2v_fallback(): # the seed is a stub path; this test is about ROLE adapter.action_submit(payload, session_factory=lambda: stub) params = stub.submitted[0]["params"] roles = [m["role"] for m in params["medias"]] _check("video" not in roles, f"a keyframe relay seed must NOT become a video media; roles={roles}") _check("<<>>" not in params["prompt"], "a relay seed must not be cited as a voice media") print("PASS test_chain_relay_seed_is_not_treated_as_voice") def test_split_references_defaults_to_todays_behaviour(): """No referenceRole -> byte-identical to before this feature existed.""" refs = ["a.png", "b.mp4"] starts, voices, warn = adapter.split_references(refs, None) _check(starts == refs and voices == [] and warn is None, f"absent role must send everything down the old path; got {starts},{voices}") starts, voices, warn = adapter.split_references(refs, "keyframe") _check(voices == [], "keyframe must never produce a voice media") starts, voices, warn = adapter.split_references(refs, "character") _check(starts == ["a.png"] and voices == ["b.mp4"] and warn is None, f"character with ONE video must split cleanly; got {starts},{voices}") print("PASS test_split_references_defaults_to_todays_behaviour") def test_autochain_plus_identity_refs_does_not_eat_the_chain_seed(): """THE real auto-chain guard. The previous one tested the wrong branch. It set referenceRole='keyframe' — but videoclaw resolves the role as `identityRefs.length ? 'character' : chainSeed ? 'keyframe' : ...`, so identity refs WIN. The configuration where the chain actually breaks therefore carries role='character', and the old test passed with the bug live. A show-bible project on --auto-chain emits role='character' with TWO videos: the previous scene's output AND the voice clip. Treating every video as a voice uploads the chain seed as a voice reference and breaks continuity invisibly. """ refs = ["scene-0.mp4", "bible-sheet.png", "voice.mp4"] starts, voices, warn = adapter.split_references(refs, "character") _check("scene-0.mp4" not in voices, f"the chain seed must NEVER be uploaded as a voice; voices={voices}") _check(voices == [], f"ambiguous input must attach NO voice rather than guess; got {voices}") _check(warn and "cannot tell" in warn, f"the ambiguity must be stated, not silent; got {warn!r}") _check(starts == refs, "refs must fall back to the pre-existing start-frame path") print("PASS test_autochain_plus_identity_refs_does_not_eat_the_chain_seed") def test_a_possibly_dropped_voice_is_never_silent(): """The old code discarded a voice on a non-character role with NO issue. That is the invisible failure this module exists to prevent: the clip renders, it is simply mute, and nothing in any status field says why. """ starts, voices, warn = adapter.split_references( ["scene-0.mp4", "voice.mp4"], "keyframe") _check(voices == [], "keyframe attaches no voice") _check(warn and "NOT applied" in warn, f"a possibly-dropped voice must be surfaced; got {warn!r}") print("PASS test_a_possibly_dropped_voice_is_never_silent") def test_media_entries_match_the_captured_submit(): """Reproduce the captured medias[]/citation shape exactly. This is the grammar higgs-cloak documented and could never exercise, because it had no way to produce a video media. Pinning it to the captured bytes so a later refactor cannot quietly drift off-contract. """ video = {"id": "6e73ff05", "type": "video_input", "url": "https://cdn/x.mp4", "width": 720, "height": 1280, "duration": 8} medias, tokens = adapter.build_media_entries(videos=[video]) _check(medias == [{"role": "video", "data": video}], f"must be role=video with the FULL media as data; got {medias}") _check(tokens == ["<<>>"], f"must cite <<>>; got {tokens}") # data is the whole object, not a trimmed triple — we do not know which # fields the server reads, so dropping any is a guess. _check(medias[0]["data"].get("duration") == 8, "data must keep the full media object, not a trimmed {id,url,type}") prompt = adapter.cite_in_prompt("<<>> in the concert singing", tokens) _check(prompt.endswith("<<>>"), f"citation must land in the prompt; got {prompt}") _check("<<>>" in prompt, "the character element citation must survive") print("PASS test_media_entries_match_the_captured_submit") def test_operator_placed_citation_is_not_duplicated(): """The captured prompt cites INLINE ('singing <<>>'), not trailing. Placement changes how the model reads it, so an operator who positioned the token deliberately must keep that position — and must not get a second copy stapled to the end. """ tokens = ["<<>>"] inline = "<<>> in the concert singing <<>> under red light" out = adapter.cite_in_prompt(inline, tokens) _check(out == inline, f"an already-cited prompt must be untouched; got {out}") _check(out.count("<<>>") == 1, "must not duplicate the citation") print("PASS test_operator_placed_citation_is_not_duplicated") def test_uncited_media_never_ships_silently(): """A media in medias[] with no citation is accepted and IGNORED. That failure is invisible: the submit succeeds, the clip does nothing, and it looks like the voice simply did not take. So an absent token gets appended. """ medias, tokens = adapter.build_media_entries(videos=[{"id": "v", "type": "video_input"}]) out = adapter.cite_in_prompt("a wide shot of the stage", tokens) _check("<<>>" in out, f"an uncited video must be cited rather than silently ignored; got {out}") print("PASS test_uncited_media_never_ships_silently") def test_no_references_yields_empty_medias(): """Text-to-video must stay byte-identical to today.""" medias, tokens = adapter.build_media_entries() _check(medias == [] and tokens == [], "no refs must produce no medias and no tokens") _check(adapter.cite_in_prompt("a plain prompt", []) == "a plain prompt", "a prompt with no tokens must be unchanged") print("PASS test_no_references_yields_empty_medias") def test_human_image_slot_compiles_to_higgsfield_token(): prompt = "ACTIVE REFERENCES: @image1 is the locked start frame. Animate @image1 only." out = adapter.compile_media_citations(prompt, ["<<>>"]) _check("@image1" not in out, f"portable slot leaked into provider prompt: {out}") _check(out.count("<<>>") == 2, f"each authored image citation should compile in place: {out}") print("PASS test_human_image_slot_compiles_to_higgsfield_token") def test_attached_start_frame_is_always_cited(): with tempfile.TemporaryDirectory() as d: frame = os.path.join(d, "start.png") with open(frame, "wb") as f: f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 64) payload = _payload(d, n_scenes=1, refs=[frame]) payload["tasks"][0]["prompt"] = "Animate the locked start frame." stub = StubSession() adapter.action_submit(payload, session_factory=lambda: stub) submitted = stub.submitted[0]["params"] _check("<<>>" in submitted["prompt"], f"attached start frame was uncited: {submitted['prompt']}") _check(submitted["medias"][0]["role"] == "image", "start media role drifted") print("PASS test_attached_start_frame_is_always_cited") def test_multiple_image_references_keep_positional_order_and_citations(): with tempfile.TemporaryDirectory() as d: refs = [] for name in ("shot.png", "character.png", "world.png"): path = os.path.join(d, name) with open(path, "wb") as f: f.write(b"\x89PNG\r\n\x1a\n" + bytes(name, "utf-8")) refs.append(path) payload = _payload(d, n_scenes=1, refs=refs) payload["tasks"][0]["prompt"] = ( "@image1 is frame zero. @image2 locks identity. @image3 locks geography." ) stub = StubSession() result = adapter.action_submit(payload, session_factory=lambda: stub) params = stub.submitted[0]["params"] _check(len(params["medias"]) == 3, f"expected 3 image medias: {params['medias']}") _check(all(m["role"] == "image" for m in params["medias"]), "image role drifted") for index in (1, 2, 3): _check(f"<<>>" in params["prompt"], f"@image{index} did not compile: {params['prompt']}") state = json.load(open(adapter.job_state_path(d, result["externalJobId"]))) applied = state["scenes"][0]["appliedImages"] _check([item["slot"] for item in applied] == ["@image1", "@image2", "@image3"], f"positional receipt drifted: {applied}") response_applied = result["rawResult"]["responses"][0]["appliedImages"] _check([item["slot"] for item in response_applied] == ["@image1", "@image2", "@image3"], f"submit response lost positional receipt: {response_applied}") print("PASS test_multiple_image_references_keep_positional_order_and_citations") def test_unreadable_wallet_refuses_to_submit_blind(): """A guard that silently stops guarding is worse than no guard. wallet() is built on api(), which NEVER throws — it returns {"status": 0, "body": None} after its retries. So a wallet outage or a response-shape change yields credits_balance None, and the old `if cb_start is not None` SKIPPED the entire spend check: full-speed submits, zero protection, and a log identical to a healthy run. """ class BlindWallet(StubSession): def wallet(self): return {"credits_balance": None, "subscription_balance": 50} with tempfile.TemporaryDirectory() as d: sess = BlindWallet() raised = None try: adapter.action_submit(_payload(d, n_scenes=5), session_factory=lambda: sess) except RuntimeError as e: raised = e _check(raised is not None, "an unreadable wallet must abort, not proceed blind") _check("credits_balance" in str(raised), f"the abort must name the cause; got {raised}") _check(len(sess.submitted) <= 1, f"must not submit the whole batch unguarded; got {len(sess.submitted)}") print("PASS test_unreadable_wallet_refuses_to_submit_blind") def test_refused_model_aborts_instead_of_exiting_zero(): """A refused model used to fail every scene and still return normally. submit_scene returns ("REFUSED: ..."), which contains no NO_UNLIM_MARKER, so the loop ran all N and action_submit returned an externalJobId with every scene failed — an unattended batch that "succeeded" with zero video. """ class RefusingSession(StubSession): def submit_scene(self, params): self.submitted.append({"params": params}) return None, "REFUSED: model 'seedance_9_0' is not on the free allowlist" with tempfile.TemporaryDirectory() as d: sess = RefusingSession() raised = None try: adapter.action_submit(_payload(d, n_scenes=8), session_factory=lambda: sess) except RuntimeError as e: raised = e _check(raised is not None, "a refused model must abort the batch") _check("refused" in str(raised).lower(), f"must name the refusal; got {raised}") _check(len(sess.submitted) == 1, f"must stop after the FIRST refusal, not run all 8; got {len(sess.submitted)}") print("PASS test_refused_model_aborts_instead_of_exiting_zero") def test_min_edge_floor_survives_non_int_dimensions(): """The floor was gated on isinstance(int). A server returning 720.0 or "720" skipped the size check silently — the same fail-open shape as the wallet guard. """ with tempfile.TemporaryDirectory() as d: for w, h, label in ((512.0, 512.0, "floats"), ("512", "512", "strings")): sess = _FakeVideoSession(CAPTURED_CREATE, finalize_body={"width": w, "height": h, "duration": 8.0}) media, reason = adapter.HiggsSession.upload_video(sess, _video_file(d)) _check(media is None, f"a sub-640 video expressed as {label} must still be refused") _check("512x512" in reason, f"reason must name the size; got {reason}") print("PASS test_min_edge_floor_survives_non_int_dimensions") def test_paid_model_refused_before_any_request(): """A paid model must be refused BEFORE the request is built. The wallet comparison in action_submit is forensic: it reads the balance, submits every scene, reads it again, and only then aborts — so on a paid model it reports the charge instead of preventing it. `HIGGS_VCLAW_I2V_MODEL` is an env var, so 'just point it at 2.0' is one typo away. The allowlist is the only check that runs early enough to matter. """ class RecordingSession(StubSession): def __init__(self): super().__init__() self.posted = [] def api(self, method, path, body=None): # must never be reached self.posted.append(path) return {"status": 200, "body": {}} sess = RecordingSession() jsid, err = adapter.HiggsSession.submit_scene(sess, {"model": "seedance_2_0", "prompt": "x"}) _check(jsid is None, "a paid model must not yield a job id") _check(err and "REFUSED" in err and "seedance_2_0" in err, f"refusal must name the model; got {err!r}") _check(sess.posted == [], f"nothing may be sent for a paid model; got {sess.posted}") # And the free ones still pass the gate. for m in ("seedance_mini_unlimited", "seedance_unlimited"): _check(m in adapter.FREE_MODELS, f"{m} must stay on the free allowlist") print("PASS test_paid_model_refused_before_any_request") def test_spend_aborts_after_one_scene_not_all(): """A charge must stop the run at the scene it happened on. The end-of-run check cannot do this — by the time it runs, all N scenes are away. With 5 scenes and a drop on the first, exactly one may be submitted. """ class SpendOnFirst(StubSession): def __init__(self): super().__init__() self._calls = 0 def wallet(self): self._calls += 1 return {"credits_balance": 100 if self._calls == 1 else 90, "subscription_balance": 50} with tempfile.TemporaryDirectory() as d: sess = SpendOnFirst() raised = None try: adapter.action_submit(_payload(d, n_scenes=5), session_factory=lambda: sess) except RuntimeError as e: raised = e _check(raised is not None, "a credits drop must abort") _check(len(sess.submitted) == 1, f"must stop after the FIRST scene, submitted {len(sess.submitted)}") _check("1 of 5" in str(raised), f"abort must say how far it got; got {raised}") jobs_dir = adapter.job_state_dir(d) _check(os.path.isdir(jobs_dir) and os.listdir(jobs_dir), "per-scene abort must still persist forensic job-state") print("PASS test_spend_aborts_after_one_scene_not_all") def test_missing_unlim_entitlement_stops_the_batch(): """No allowance -> every scene fails identically; don't grind through them. Each failing submit burns a full retry budget, so 25 scenes cost 25x that for an answer the first scene already gave. """ class NoUnlim(StubSession): def submit_scene(self, params): return None, ('HTTP 403 from POST /jobs/v2/seedance_unlimited: ' '{"detail": {"error_type": "unlimited_generation_not_allowed"}}') with tempfile.TemporaryDirectory() as d: sess = NoUnlim() raised = None try: adapter.action_submit(_payload(d, n_scenes=6), session_factory=lambda: sess) except RuntimeError as e: raised = e _check(raised is not None and "unlimited_generation_not_allowed" in str(raised), f"must abort naming the entitlement; got {raised}") _check("1 of 6" in str(raised), f"must stop after the first; got {raised}") _check("resumable" in str(raised), "must tell the operator the queue resumes") print("PASS test_missing_unlim_entitlement_stops_the_batch") # --------------------------------------------------------------------------- # Test 5: param mapping (executionProfile -> Higgsfield free submit shape). # --------------------------------------------------------------------------- def test_param_mapping(): p = adapter.build_params( {"aspectRatio": "9:16", "resolution": "1080p", "generateAudio": False}, "hello", 12) # Was pinned to the literal "seedance_mini_unlimited". That model now answers # unlimited_generation_not_allowed — the unlimited path moved onto the # use_unlim flag over a current model — so pinning the NAME would fail the # build every time the provider rotates its free tier. The property actually # worth guarding is that the default can never be a paid model. _check(p["model"] in adapter.FREE_MODELS, f"default model must be on the free allowlist; got {p['model']!r}") _check(p["aspect_ratio"] == "9:16", "aspect mapped") # The free endpoint 422s on anything but 480p/720p — a 1080p profile must # CLAMP to 720p, not pass through (the Last Call first-night failure). _check(p["resolution"] == "720p", "1080p clamps to 720p (free endpoint limit)") _check((p["width"], p["height"]) == (720, 1280), "9:16 dims follow the clamped resolution") p720 = adapter.build_params({"aspectRatio": "16:9", "resolution": "480p"}, "x", 4) _check(p720["resolution"] == "480p", "supported resolutions pass through unclamped") _check(p["generate_audio"] is False, "generateAudio mapped") _check(p["duration"] == 12, "durationSeconds mapped") _check(p["medias"] == [], "medias empty for text-to-video (no start frame)") body = adapter.submit_body(p) _check(body["use_unlim"] is True and body["use_free_gens"] is False, "free path flags pinned") # i2v variant: a start-frame media + the proven i2v model, free flags unchanged. media = {"role": "image", "data": {"type": "media_input", "url": "u", "id": "m"}} p2 = adapter.build_params({"aspectRatio": "16:9"}, "hi", 8, medias=[media], model=adapter.MODEL_I2V) _check(p2["model"] == adapter.MODEL_I2V, "i2v routes to MODEL_I2V") _check(p2["medias"] == [media], "i2v carries the uploaded start frame") b2 = adapter.submit_body(p2) _check(b2["use_unlim"] is True and b2["use_free_gens"] is False, "i2v stays on the free path (use_unlim/use_free_gens unchanged)") print("PASS test_param_mapping") # --------------------------------------------------------------------------- # Test 8: i2v — an image referencePath is uploaded + attached as the start frame. # --------------------------------------------------------------------------- def test_i2v_image_reference_uploaded(): with tempfile.TemporaryDirectory() as d: img = os.path.join(d, "start.png") with open(img, "wb") as f: # a real file on disk (content unchecked here) f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 64) stub = StubSession(poll_status="completed") sub = adapter.action_submit(_payload(d, n_scenes=2, refs=[img]), session_factory=lambda: stub) # The start frame was uploaded exactly once (scene 0 has the ref; scene 1 none). _check(len(stub.uploads) == 1, f"expected 1 upload, got {len(stub.uploads)}") _check(stub.uploads[0]["existed"], "the uploaded start frame must exist on disk") # Scene 0 submitted as image-to-video with the media attached; scene 1 T2V. p0 = stub.submitted[0]["params"] _check(p0["model"] == adapter.MODEL_I2V, "scene 0 routed to the i2v model") _check(len(p0["medias"]) == 1 and p0["medias"][0]["role"] == "image", "scene 0 carries the start-frame media") _check(p0["medias"][0]["data"]["type"] == "media_input", "media_input shape") p1 = stub.submitted[1]["params"] _check(p1["model"] == adapter.MODEL and p1["medias"] == [], "scene 1 (no ref) stays byte-identical text-to-video") # No spurious 'not applied' issue for the applied scene. _check(not any("not applied" in i for i in sub["rawResult"]["issues"]), "an applied start frame must not raise a 'not applied' issue") # Job-state records provenance for a later poll. state = json.load(open(adapter.job_state_path(d, sub["externalJobId"]))) s0 = state["scenes"][0] _check(s0["referenceApplied"] is True, "scene 0 referenceApplied=True") _check(s0["startFrame"] and s0["startFrame"]["mediaId"], "scene 0 startFrame records the uploaded mediaId") _check(s0["startFrame"]["extractedFromVideo"] is False, "image ref not from video") _check(state["scenes"][1]["referenceApplied"] is False, "scene 1 not i2v") # POLL must NOT re-surface a 'not applied' advisory for the applied scene. poll = adapter.action_poll( {"action": "poll", "outputDir": d, "externalJobId": sub["externalJobId"], "workspaceRoot": d}, session_factory=lambda: StubSession(poll_status="completed")) _check(not any("not applied" in i for i in poll["issues"]), "poll must not warn 'not applied' for a scene that applied its start frame") print("PASS test_i2v_image_reference_uploaded") # --------------------------------------------------------------------------- # Test 9: relay — a VIDEO referencePath yields a tail-frame PNG (needs ffmpeg). # --------------------------------------------------------------------------- def test_relay_video_tail_frame(): import shutil as _sh if not _sh.which("ffmpeg"): print("SKIP test_relay_video_tail_frame (ffmpeg not installed)") return with tempfile.TemporaryDirectory() as d: vid = os.path.join(d, "prev-scene.mp4") # A 1s color clip is enough for a decodable last frame. r = subprocess.run( ["ffmpeg", "-nostdin", "-y", "-f", "lavfi", "-i", "testsrc2=size=320x240:rate=12:duration=1", vid], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) _check(r.returncode == 0 and os.path.exists(vid), "test video generated") tmp = os.path.join(d, "frames") os.makedirs(tmp, exist_ok=True) img, mime, source = adapter.resolve_start_frame([vid], tmp) _check(img is not None and os.path.exists(img), "a video referencePath must yield an extracted tail frame") _check(mime == "image/png", "extracted tail frame is a PNG") _check(source["extractedFromVideo"] is True, "provenance marks a video extraction") _check(os.path.getsize(img) > 0, "extracted frame is non-empty") # And through submit: the tail frame is uploaded and the scene is i2v. stub = StubSession() sub = adapter.action_submit(_payload(d, n_scenes=1, refs=[vid]), session_factory=lambda: stub) _check(len(stub.uploads) == 1 and stub.uploads[0]["path"].endswith(".png"), "the relay tail frame (a .png) was uploaded") state = json.load(open(adapter.job_state_path(d, sub["externalJobId"]))) _check(state["scenes"][0]["startFrame"]["extractedFromVideo"] is True, "job-state marks the relay tail-frame provenance") print("PASS test_relay_video_tail_frame") # --------------------------------------------------------------------------- # Test 10: audio-only / missing refs degrade to T2V (never block, never spend). # --------------------------------------------------------------------------- def test_unusable_reference_degrades_to_t2v(): with tempfile.TemporaryDirectory() as d: aud = os.path.join(d, "voice.mp3") with open(aud, "wb") as f: f.write(b"ID3fake") stub = StubSession() with allow_t2v_fallback(): sub = adapter.action_submit(_payload(d, n_scenes=1, refs=[aud, "/does/not/exist.png"]), session_factory=lambda: stub) _check(len(stub.uploads) == 0, "no upload for audio/missing refs") p0 = stub.submitted[0]["params"] _check(p0["model"] == adapter.MODEL and p0["medias"] == [], "unusable refs render byte-identical text-to-video") _check(any("text-to-video" in i for i in sub["rawResult"]["issues"]), "a non-fatal advisory explains the T2V fallback") print("PASS test_unusable_reference_degrades_to_t2v") # --------------------------------------------------------------------------- # A REQUESTED start frame that cannot be applied must abort the submit, not # quietly render text-to-video. Dropping the identity anchor produces a clip # that passes every automated check while showing the wrong subject — on a real # film it put generic toddlers into a Krishna episode, and nothing downstream # could see it. The degrade path survives only behind an explicit env var. # --------------------------------------------------------------------------- def test_missing_start_frame_refuses_by_default(): with tempfile.TemporaryDirectory() as d: stub = StubSession() raised = None try: adapter.action_submit(_payload(d, n_scenes=1, refs=["/does/not/exist.png"]), session_factory=lambda: stub) except RuntimeError as e: raised = e _check(raised is not None, "a start frame that cannot be applied must raise, not degrade") _check("REFUSING" in str(raised), f"the refusal must say so; got {raised}") _check(len(stub.submitted) == 0, "nothing may be submitted once the identity anchor is known lost") # ...and the escape hatch still works, so an operator who wants T2V can have it. with allow_t2v_fallback(): sub = adapter.action_submit(_payload(d, n_scenes=1, refs=["/does/not/exist.png"]), session_factory=lambda: stub) _check(any("text-to-video" in i for i in sub["rawResult"]["issues"]), "the opt-in path still degrades and still says so") print("PASS test_missing_start_frame_refuses_by_default") # --------------------------------------------------------------------------- # Test 12: a start-frame rejected by moderation must NOT be handed to job # creation (which 404s "Media input not found"); the scene degrades to T2V and # the real moderation reason is surfaced. Regression for the swallowed # finalize-422 / terminal-nsfw bug that submitted a dead media. # --------------------------------------------------------------------------- def test_moderation_rejected_reference_degrades_with_reason(): class RejectingUploadSession(StubSession): REASON = ("reference image rejected at upload finalize (HTTP 422: " '{"detail": {"error_type": "nsfw", "text": "Restricted content detected"}})') def upload_image(self, path, mime="image/png"): self.uploads.append({"path": path, "mime": mime}) return None, self.REASON with tempfile.TemporaryDirectory() as d: img = os.path.join(d, "anchor.png") with open(img, "wb") as f: f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 64) stub = RejectingUploadSession() with allow_t2v_fallback(): sub = adapter.action_submit(_payload(d, n_scenes=1, refs=[img]), session_factory=lambda: stub) # The moderation reason reaches issues[] (not a bare "upload failed"). _check(any(RejectingUploadSession.REASON in i for i in sub["rawResult"]["issues"]), f"moderation reason must surface; got {sub['rawResult']['issues']}") _check(any("identity anchor dropped" in i for i in sub["rawResult"]["issues"]), "operator must be told the identity anchor was dropped") # The rejected media was NOT attached; the scene fell back to T2V and still # got a (stub) job_set id — no dead-media job creation. p0 = stub.submitted[0]["params"] _check(p0["model"] == adapter.MODEL and p0["medias"] == [], "rejected reference must degrade to byte-identical T2V") state = json.load(open(adapter.job_state_path(d, sub["externalJobId"]))) _check(state["scenes"][0]["referenceApplied"] is False, "job-state must record the reference was not applied") _check(state["scenes"][0]["status"] == "submitted", "the T2V fallback scene still submitted") print("PASS test_moderation_rejected_reference_degrades_with_reason") # --------------------------------------------------------------------------- # Test 11: a job-creation rejection surfaces the server's reason (not a bare # "no job_set id returned"). Regression for the swallowed-error bug that made a # moderation/validation gate at job creation indistinguishable from a dead run. # --------------------------------------------------------------------------- def test_submit_failure_surfaces_server_reason(): class RejectingSession(StubSession): REASON = 'HTTP 422 from POST /jobs/v2/seedance_unlimited: {"detail":"content policy: minor"}' def submit_scene(self, params): self.submitted.append({"jsid": None, "params": params}) return None, self.REASON with tempfile.TemporaryDirectory() as d: stub = RejectingSession() sub = adapter.action_submit(_payload(d, n_scenes=1), session_factory=lambda: stub) # Top-level submittedScenes still carries the (empty) taskId contract. _check(sub["rawResult"]["submittedScenes"][0]["taskId"] == "", "a failed submit keeps an empty taskId") # The real reason reaches issues[] AND the per-response submitError. _check(any(RejectingSession.REASON in i for i in sub["rawResult"]["issues"]), f"server reason must reach issues[]; got {sub['rawResult']['issues']}") _check(sub["rawResult"]["responses"][0].get("submitError") == RejectingSession.REASON, "per-scene response must carry submitError") # Persisted job-state error carries the reason for the later poll to re-surface. state = json.load(open(adapter.job_state_path(d, sub["externalJobId"]))) s0 = state["scenes"][0] _check(s0["status"] == "failed", "rejected scene is failed") _check(RejectingSession.REASON in s0.get("error", ""), f"job-state error must carry the reason; got {s0.get('error')}") # POLL re-surfaces that stored reason (videoclaw reads the poll issues[]). poll = adapter.action_poll( {"action": "poll", "outputDir": d, "externalJobId": sub["externalJobId"], "workspaceRoot": d}, session_factory=lambda: StubSession()) _check(poll["status"] == "failed", "a no-taskId scene polls as failed") _check(any(RejectingSession.REASON in i for i in poll["issues"]), f"poll must re-surface the stored reason; got {poll['issues']}") print("PASS test_submit_failure_surfaces_server_reason") # --------------------------------------------------------------------------- # Test 6: subprocess transport through the REAL wrapper (stdin -> stdout JSON). # --------------------------------------------------------------------------- def test_subprocess_transport(): """Drive the adapter exactly as videoclaw does: spawn the wrapper, pipe a SUBMIT payload on stdin, parse one JSON object from stdout. The adapter's make_session is swapped for a stub via HIGGS_VCLAW_TEST_STUB so no browser launches.""" with tempfile.TemporaryDirectory() as d: payload = _payload(d, n_scenes=1) env = dict(os.environ, HIGGS_VCLAW_TEST_STUB="1", PYTHONPATH=HERE) # Use the wrapper script with a --route arg, exactly like the env var. proc = subprocess.run( ["/bin/sh", os.path.join(HERE, "run.sh"), "--route", "seedance-direct"], input=json.dumps(payload), capture_output=True, text=True, env=env, ) _check(proc.returncode == 0, f"adapter exited {proc.returncode}; stderr={proc.stderr[:500]}") # CONTRACT: stdout must be CLEAN JSON ONLY — exactly one line that parses. # The stub prints a fake-browser banner to stdout; the guard must route # it to stderr so the line below is pure JSON. lines = [ln for ln in proc.stdout.splitlines() if ln.strip()] _check(len(lines) == 1, f"stdout must be exactly one JSON line; got {len(lines)}: {proc.stdout[:300]}") out = json.loads(lines[0]) _check(isinstance(out.get("externalJobId"), str) and out["externalJobId"], f"subprocess submit must print externalJobId; got {proc.stdout[:300]}") # The banner must have been redirected to stderr, not stdout. _check("must NOT reach stdout" not in proc.stdout, "session stdout-noise leaked into the JSON channel") _check("must NOT reach stdout" in proc.stderr, "session stdout-noise should appear on stderr") print("PASS test_subprocess_transport") # --------------------------------------------------------------------------- # Test 7: POLL 'completed' with empty outputs must be reported as 'failed'. # --------------------------------------------------------------------------- def test_completed_empty_outputs_is_failure(): # A session that returns a completed job carrying NO result url -> no output # can be produced, so the aggregate must be 'failed', never 'completed'. class NoUrlSession(StubSession): def query_job(self, job_set_id): job = {"id": f"job-{job_set_id}", "status": "completed", "results": {}} return job, {"jobs": [job]} with tempfile.TemporaryDirectory() as d: sub = adapter.action_submit(_payload(d, n_scenes=1), session_factory=lambda: StubSession()) eid = sub["externalJobId"] # No url + status 'completed' is not terminal/fail -> stays pending here, # which is correct (the clip simply isn't ready). Verify it is NOT # reported completed-with-empty-outputs. res = adapter.action_poll( {"action": "poll", "outputDir": d, "externalJobId": eid, "workspaceRoot": d}, session_factory=lambda: NoUrlSession()) _check(res["status"] != "completed" or res["outputs"], "must never return status=completed with empty outputs") print("PASS test_completed_empty_outputs_is_failure") def test_relative_reference_resolves_against_project_dir(): """A PROJECT-RELATIVE referencePath (how videoclaw asset manifests store paths) must resolve against workspaceRoot/projects/ and still drive i2v — before this, it silently degraded the scene to text-to-video (the Last Call absolute-path foot-gun).""" with tempfile.TemporaryDirectory() as d: kf_dir = os.path.join(d, "projects", "selftest", "references") os.makedirs(kf_dir) with open(os.path.join(kf_dir, "kf.png"), "wb") as f: f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 64) stub = StubSession(poll_status="completed") adapter.action_submit(_payload(d, n_scenes=1, refs=["references/kf.png"]), session_factory=lambda: stub) _check(len(stub.uploads) == 1, f"relative ref must upload a start frame, got {len(stub.uploads)} uploads") _check(stub.uploads[0]["existed"], "the resolved relative ref must exist on disk") _check(stub.submitted[0]["params"]["model"] == adapter.MODEL_I2V, "relative ref still routes the scene to the i2v model") print("PASS test_relative_reference_resolves_against_project_dir") def test_extra_params_default_off_is_byte_identical(): """No override -> the submitted params carry none of the passthrough keys. 2.5 already renders without any of them, so the default MUST add nothing; the passthrough exists to exercise the fields, not because they are needed. """ import subprocess, json as _json, os as _os, sys as _sys code = ( "import sys, json; sys.path.insert(0, %r); import adapter;" "print(json.dumps(adapter.build_params({'aspectRatio':'16:9','resolution':'720p'}, 'p', 8), sort_keys=True))" % _os.path.dirname(_os.path.abspath(__file__)) ) env = {**_os.environ}; env.pop("HIGGS_VCLAW_EXTRA_PARAMS", None) out = subprocess.run([_sys.executable, "-c", code], capture_output=True, text=True, env=env) params = _json.loads(out.stdout) for k in ("genre", "multi_shots", "multi_shot_mode", "multi_prompt", "speedramp", "reference_elements", "prompt_language", "extension_mode"): _check(k not in params, f"default submit must not carry {k}") print("PASS test_extra_params_default_off_is_byte_identical") def test_extra_params_land_but_never_clobber_the_submit_identity(): import subprocess, json as _json, os as _os, sys as _sys code = ( "import sys, json; sys.path.insert(0, %r); import adapter;" "print(json.dumps(adapter.build_params({'aspectRatio':'16:9','resolution':'720p'}, 'REAL', 8), sort_keys=True))" % _os.path.dirname(_os.path.abspath(__file__)) ) env = {**_os.environ, "HIGGS_VCLAW_EXTRA_PARAMS": _json.dumps({ "genre": "auto", "multi_shots": True, "multi_shot_mode": "custom", "prompt_language": "en", # These three must be ignored: they are the identity of the submit. "model": "hijacked", "prompt": "hijacked", "medias": ["hijacked"], })} out = subprocess.run([_sys.executable, "-c", code], capture_output=True, text=True, env=env) params = _json.loads(out.stdout) _check(params.get("multi_shots") is True, "multi_shots should pass through") _check(params.get("multi_shot_mode") == "custom", "multi_shot_mode should pass through") _check(params["prompt"] == "REAL", "prompt must not be overridable") _check(params["model"] != "hijacked", "model must not be overridable") _check(params["medias"] == [], "medias must not be overridable") print("PASS test_extra_params_land_but_never_clobber_the_submit_identity") def test_malformed_extra_params_fails_loudly(): """A malformed override must abort, not be silently ignored — otherwise a typo submits a different job than the operator asked for.""" import subprocess, os as _os, sys as _sys code = "import sys; sys.path.insert(0, %r); import adapter" % _os.path.dirname(_os.path.abspath(__file__)) env = {**_os.environ, "HIGGS_VCLAW_EXTRA_PARAMS": "{not json"} out = subprocess.run([_sys.executable, "-c", code], capture_output=True, text=True, env=env) _check(out.returncode != 0, "malformed extra params must abort") _check("not a valid JSON object" in (out.stderr + out.stdout), "abort message should name the cause") print("PASS test_malformed_extra_params_fails_loudly") def test_profile_path_is_relocatable_and_overrideable(): """The live adapter must never inherit a developer-specific absolute path.""" expected = os.path.join(HERE, ".cloak-profile") _check(adapter.DEFAULT_PROFILE == expected, f"default profile must live beside the engine; got {adapter.DEFAULT_PROFILE!r}") custom = os.path.join(tempfile.gettempdir(), "higgs-profile-override-test") code = ( "import sys; sys.path.insert(0, %r); import adapter; " "print(adapter.DEFAULT_PROFILE)" % HERE ) env = {**os.environ, "HIGGS_VCLAW_PROFILE": custom} out = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, env=env) _check(out.returncode == 0, f"profile override import failed: {out.stderr}") _check(out.stdout.strip() == custom, f"HIGGS_VCLAW_PROFILE override ignored; got {out.stdout.strip()!r}") print("PASS test_profile_path_is_relocatable_and_overrideable") def test_motion_reference_rides_as_video_before_the_voice(): """task.motionReferencePath (a muted donor clip) must land as a `video` media cited <<>>, BEFORE the voice (`audio_1`), with the voice still attached — and it must go through upload_video, never through referencePaths (where a second video would silently drop the voice).""" with tempfile.TemporaryDirectory() as d: img = os.path.join(d, "keyframe.png") with open(img, "wb") as f: f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 64) voice = os.path.join(d, "w05.mp4") with open(voice, "wb") as f: f.write(b"\x00\x00\x00 ftypisom" + b"\x00" * 256) donor = os.path.join(d, "w05-donor.mp4") with open(donor, "wb") as f: f.write(b"\x00\x00\x00 ftypisom" + b"\x01" * 256) payload = _payload(d, n_scenes=1) payload["tasks"][0]["referencePaths"] = [img, voice] payload["tasks"][0]["referenceRole"] = "character" payload["tasks"][0]["motionReferencePath"] = donor payload["tasks"][0]["prompt"] = "she copies the moves of @video1 on the sun disc" stub = StubSession() with allow_t2v_fallback(): out = adapter.action_submit(payload, session_factory=lambda: stub) params = stub.submitted[0]["params"] roles = [m["role"] for m in params["medias"]] _check(roles == ["image", "video", "audio"] or roles == ["image", "video", "video"], f"expected image, video(donor), then the voice; roles={roles}") _check("<<>>" in params["prompt"], f"the donor must be cited: {params['prompt']!r}") _check("@video1" not in params["prompt"], "the human @video1 token must be compiled to <<>>") donor_uploads = [u for u in stub.uploads if u["path"] == donor] _check(donor_uploads and donor_uploads[0]["mime"].startswith("video/"), f"the donor must go through upload_video; uploads={stub.uploads}") scene = out["scenes"][0] if isinstance(out, dict) and out.get("scenes") else None if scene is not None: _check((scene.get("appliedMotion") or {}).get("slot") == "@video1", f"the receipt must record appliedMotion; got {scene.get('appliedMotion')}") print("PASS test_motion_reference_rides_as_video_before_the_voice") def test_no_motion_reference_is_byte_identical(): """The field is optional: a payload without it builds exactly the params it built before the donor existed.""" with tempfile.TemporaryDirectory() as d: img = os.path.join(d, "portrait.jpg") with open(img, "wb") as f: f.write(b"\xff\xd8\xff" + b"\x00" * 64) voice = os.path.join(d, "w08.mp4") with open(voice, "wb") as f: f.write(b"\x00\x00\x00 ftypisom" + b"\x00" * 256) payload = _payload(d, n_scenes=1) payload["tasks"][0]["referencePaths"] = [img, voice] payload["tasks"][0]["referenceRole"] = "character" payload["tasks"][0]["prompt"] = "the performer on a neon rooftop" a, b = StubSession(), StubSession() with allow_t2v_fallback(): adapter.action_submit(json.loads(json.dumps(payload)), session_factory=lambda: a) p2 = json.loads(json.dumps(payload)); p2["tasks"][0].pop("motionReferencePath", None) adapter.action_submit(p2, session_factory=lambda: b) _check(a.submitted[0]["params"] == b.submitted[0]["params"], "a payload with no motionReferencePath must build identical params") _check(len(a.submitted[0]["params"]["medias"]) == 2, "no donor → exactly the still and the voice (the stub's voice rides as video: no ffmpeg on a fake file)") print("PASS test_no_motion_reference_is_byte_identical") def test_refused_motion_reference_aborts(): """A donor the media gate refuses is a HARD FAIL — the take would look valid and copy no moves. HIGGS_VCLAW_ALLOW_NO_MOTION=1 is the only way to render without it.""" class Refusing(StubSession): def upload_video(self, path, mime="video/mp4"): self.uploads.append({"path": path, "mime": mime, "existed": os.path.exists(path)}) return (None, "rejected: ip") with tempfile.TemporaryDirectory() as d: img = os.path.join(d, "keyframe.png") with open(img, "wb") as f: f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 64) donor = os.path.join(d, "w05-donor.mp4") with open(donor, "wb") as f: f.write(b"\x00\x00\x00 ftypisom" + b"\x01" * 256) payload = _payload(d, n_scenes=1) payload["tasks"][0]["referencePaths"] = [img] payload["tasks"][0]["referenceRole"] = "character" payload["tasks"][0]["motionReferencePath"] = donor payload["tasks"][0]["prompt"] = "she copies @video1" stub = Refusing() raised = None os.environ.pop("HIGGS_VCLAW_ALLOW_NO_MOTION", None) try: with allow_t2v_fallback(): adapter.action_submit(payload, session_factory=lambda: stub) except RuntimeError as exc: raised = str(exc) _check(raised is not None and "motion reference NOT applied" in raised and "w05-donor.mp4" in raised, f"a refused donor must abort naming the donor; got {raised!r}") _check(not stub.submitted, "nothing may be submitted after a refused donor") print("PASS test_refused_motion_reference_aborts") if __name__ == "__main__": test_submit_poll_roundtrip() test_poll_pending_and_failed() test_cancel() test_wallet_is_read_only_and_closes_session() test_free_safety_abort() test_api_url_targets_the_datadome_allowlisted_host() test_fetch_surfaces_bot_protection_response_headers() test_submits_are_paced_to_stay_under_turnstile() test_duration_clamps_to_the_model_window() test_submit_body_sends_the_mode_not_the_model_name() test_voice_goes_on_the_audio_channel_not_as_a_black_frame_video() test_media_entries_route_by_media_type() test_voice_clip_rides_alongside_the_start_frame() test_chain_relay_seed_is_not_treated_as_voice() test_split_references_defaults_to_todays_behaviour() test_autochain_plus_identity_refs_does_not_eat_the_chain_seed() test_a_possibly_dropped_voice_is_never_silent() test_media_entries_match_the_captured_submit() test_operator_placed_citation_is_not_duplicated() test_uncited_media_never_ships_silently() test_no_references_yields_empty_medias() test_human_image_slot_compiles_to_higgsfield_token() test_attached_start_frame_is_always_cited() test_multiple_image_references_keep_positional_order_and_citations() test_video_upload_parses_the_dict_response() test_video_upload_tolerates_a_list_response() test_upload_waits_for_the_ip_check_not_just_the_bytes() test_video_under_640_is_refused_with_the_reason() test_video_path_404_explains_the_prefix_mapping() test_video_rejected_by_moderation_surfaces_status() test_unreadable_wallet_refuses_to_submit_blind() test_refused_model_aborts_instead_of_exiting_zero() test_min_edge_floor_survives_non_int_dimensions() test_paid_model_refused_before_any_request() test_spend_aborts_after_one_scene_not_all() test_missing_unlim_entitlement_stops_the_batch() test_param_mapping() test_i2v_image_reference_uploaded() test_relay_video_tail_frame() test_missing_start_frame_refuses_by_default() test_unusable_reference_degrades_to_t2v() test_moderation_rejected_reference_degrades_with_reason() test_submit_failure_surfaces_server_reason() test_subprocess_transport() test_completed_empty_outputs_is_failure() test_relative_reference_resolves_against_project_dir() test_extra_params_default_off_is_byte_identical() test_extra_params_land_but_never_clobber_the_submit_identity() test_malformed_extra_params_fails_loudly() test_profile_path_is_relocatable_and_overrideable() test_motion_reference_rides_as_video_before_the_voice() test_no_motion_reference_is_byte_identical() test_refused_motion_reference_aborts() print("\nALL TESTS PASS")