#!/usr/bin/env python3
"""
RUBRA — scripts/test_hf_render.py
Decoupled HyperFrames verification probe.
Run this manually (or in a CI/CD pre-deploy step) AFTER the Dockerfile
installs node + hyperframes, BEFORE flipping RUBRA_VIDEO_ENABLED=true.
What it does:
1. Renders a tiny, deliberately cheap HTML scene (small viewport, ~1s,
no external assets) — enough to prove the node/hyperframes/CPU
pipeline works on THIS container, without burning a full render budget.
2. Never raises past its own boundary — any exception is caught and
reported as a clean FAIL, so this can't itself crash a CI runner.
3. Exits 0 on pass (safe to flip the env var) / 1 on fail (keep it off).
4. Optionally writes the verdict to a state file so a deploy script can
read pass/fail without parsing stdout.
This script is intentionally decoupled from main.py/agent.py — it
imports nothing from the live app, so it can be run standalone in a
throwaway container during image build/CI, before the real app ever
starts.
"""
import os
import sys
import json
import time
import shutil
import subprocess
import tempfile
from pathlib import Path
RENDER_TIMEOUT_S = 45 # generous for a tiny scene; production renders use 120s
STATE_FILE = os.getenv("RUBRA_RENDER_TEST_STATE", "/tmp/hf_render_test_result.json")
# Deliberately tiny: small canvas, ~1 second, no images/fonts/network —
# this tests the pipeline (node → hyperframes → ffmpeg-equivalent → mp4
# bytes on disk), not rendering performance under real workload.
_PROBE_HTML = """
"""
def _write_state(passed: bool, detail: str):
try:
Path(STATE_FILE).write_text(json.dumps({
"passed": passed,
"detail": detail[:500],
"timestamp": time.time(),
}))
except Exception:
pass # state file is a convenience, never a hard requirement
def check_binaries() -> tuple[bool, str]:
"""Cheap pre-check before even attempting a render."""
if shutil.which("node") is None:
return False, "node binary not found on PATH — Dockerfile node install step missing/failed"
if shutil.which("npx") is None:
return False, "npx binary not found on PATH"
return True, "node/npx present"
def run_probe_render() -> tuple[bool, str]:
"""The actual tiny render attempt. Never lets an exception escape."""
try:
with tempfile.TemporaryDirectory() as tmp:
html_path = os.path.join(tmp, "probe.html")
out_path = os.path.join(tmp, "probe.mp4")
Path(html_path).write_text(_PROBE_HTML, encoding="utf-8")
start = time.time()
result = subprocess.run(
["npx", "hyperframes", "render", html_path, "-o", out_path],
capture_output=True, text=True, timeout=RENDER_TIMEOUT_S,
)
elapsed = time.time() - start
if result.returncode != 0:
return False, f"hyperframes exited {result.returncode} after {elapsed:.1f}s: {result.stderr[:300]}"
if not os.path.exists(out_path):
return False, "hyperframes reported success but produced no output file"
size = os.path.getsize(out_path)
if size < 100: # a real MP4 container header alone is well over this
return False, f"output file exists but is suspiciously small ({size} bytes) — likely corrupt"
return True, f"Rendered {size} bytes in {elapsed:.1f}s — pipeline confirmed working"
except subprocess.TimeoutExpired:
return False, f"Render exceeded {RENDER_TIMEOUT_S}s on a TINY probe scene — this container's CPU pool likely can't sustain real renders"
except FileNotFoundError as e:
return False, f"Binary not found during execution: {e}"
except Exception as e:
# Catch-all: a CPU-pool crash, OOM kill, or any other container-level
# failure must fail SAFE here, never propagate and take down a CI step.
return False, f"Unexpected exception during probe render: {type(e).__name__}: {e}"
def main() -> int:
print("[HF_RENDER_TEST] Checking for node/npx binaries...")
bin_ok, bin_detail = check_binaries()
print(f"[HF_RENDER_TEST] {bin_detail}")
if not bin_ok:
_write_state(False, bin_detail)
print("[HF_RENDER_TEST] FAIL — keeping RUBRA_VIDEO_ENABLED=false")
return 1
print("[HF_RENDER_TEST] Running tiny probe render...")
render_ok, render_detail = run_probe_render()
print(f"[HF_RENDER_TEST] {render_detail}")
_write_state(render_ok, render_detail)
if render_ok:
print("[HF_RENDER_TEST] PASS — safe to set RUBRA_VIDEO_ENABLED=true")
return 0
else:
print("[HF_RENDER_TEST] FAIL — keeping RUBRA_VIDEO_ENABLED=false")
return 1
if __name__ == "__main__":
sys.exit(main())