| """A real-core smoke test with a deterministic local HTTP endpoint. |
| |
| No GPU/model is used. The fixture makes an explicit replay tool call; the |
| assertion checks that core actually sends the historical red frame afterward. |
| """ |
| import base64 |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
| import io |
| import json |
| import os |
| from pathlib import Path |
| import subprocess |
| import sys |
| import threading |
| import time |
| from types import SimpleNamespace |
|
|
| from PIL import Image |
| import pytest |
|
|
| from demo.config import Settings |
| from demo.runner import atomic_json |
|
|
| ROOT = Path(__file__).resolve().parents[2] |
|
|
|
|
| def make_video(path): |
| from demo.media import binary |
| frames = path.parent / "source-frames" |
| frames.mkdir() |
| for index, color in enumerate(["red", "green", "blue", "white", "black"]): |
| Image.new("RGB", (64, 48), color).save(frames / f"{index:03d}.png") |
| subprocess.run([ |
| binary("ffmpeg"), "-nostdin", "-v", "error", "-y", "-threads", "2", |
| "-framerate", "1", "-i", str(frames / "%03d.png"), |
| "-c:v", "libx264", "-threads", "2", "-pix_fmt", "yuv420p", "-r", "10", str(path), |
| ], check=True, capture_output=True) |
|
|
|
|
| @pytest.mark.parametrize("query_time,actual_time,via_server", [(3.8, 3, False), (5.0, 4, False), (3.8, 3, True)]) |
| def test_real_core_replay_sends_historical_image(tmp_path, query_time, actual_time, via_server): |
| seen = [] |
| replay_requested = False |
| replay_received = [] |
|
|
| class Handler(BaseHTTPRequestHandler): |
| def log_message(self, *args): |
| pass |
|
|
| def do_POST(self): |
| nonlocal replay_requested |
| payload = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) |
| seen.append(payload) |
| user_text = "\n".join( |
| message["content"] if isinstance(message["content"], str) |
| else "\n".join(p.get("text", "") for p in message["content"]) |
| for message in payload["messages"] if message["role"] == "user" |
| ) |
| images = [part["image_url"]["url"] for message in payload["messages"] |
| if isinstance(message["content"], list) for part in message["content"] |
| if part.get("type") == "image_url"] |
| if not images: |
| answer = "A solid red frame was shown before the later colors." |
| elif replay_requested: |
| replay_received.extend(images) |
| answer = "I rewatched the earlier frame and saw red. </Response> Red." |
| elif "[Phase] OBSERVE-ONLY." in user_text: |
| answer = ("<Observation>description: A colored background is visible; actors: none; " |
| "objects: background; actions: none; ocr: none; scene: solid background; " |
| f"delta: The visible color at this observation is recorded at call {len(seen)}." |
| "</Observation>\n</Silence>") |
| elif "PLANNING STEP" in user_text: |
| answer = ("<Plan>\nneed: Identify the earliest color\ninfo_type: color\n" |
| "evidence_locus: past frame at 0 seconds\nanswerability: need to replay\n" |
| "meta: ordinary question\nstrategy: replay 0 seconds then answer\n</Plan>\n</Silence>") |
| else: |
| replay_requested = True |
| answer = ('The current background is white, but the question asks about the first frame. I will replay t=0 to check the initial color. ' |
| '<tool_call>{"name":"replay","arguments":{"t":0}}</tool_call>') |
| body = json.dumps({"choices": [{"message": {"content": answer}}]}).encode() |
| self.send_response(200) |
| self.send_header("Content-Type", "application/json") |
| self.send_header("Content-Length", str(len(body))) |
| self.end_headers() |
| self.wfile.write(body) |
|
|
| server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) |
| thread = threading.Thread(target=server.serve_forever, daemon=True) |
| thread.start() |
| video = tmp_path / "test.mp4" |
| make_video(video) |
| job = tmp_path / "job" |
| job.mkdir() |
| atomic_json(job / "job.json", { |
| "id": "fixture", "videoId": "fixture-video", "video_path": str(video), |
| "question": "What color appeared at the beginning?", "query_time": query_time, |
| "duration": 5, "max_steps": 20, "max_model_calls": 20, |
| "enable_audio": False, "enable_tracking": False, |
| }) |
| settings = Settings( |
| data_dir=tmp_path / "server-data", api_base_url=f"http://127.0.0.1:{server.server_port}/v1", |
| model="fixture", api_key="", max_model_calls=20, |
| ) |
| env = settings.worker_env() |
| try: |
| if via_server: |
| from fastapi.testclient import TestClient |
| from demo.server import create_app |
|
|
| with TestClient(create_app(settings)) as client: |
| with video.open("rb") as handle: |
| uploaded = client.post("/api/videos", files={"video": ("test.mp4", handle, "video/mp4")}) |
| assert uploaded.status_code == 201, uploaded.text |
| started = client.post("/api/runs", json={ |
| "videoId": uploaded.json()["id"], |
| "question": "What color appeared at the beginning?", "queryTime": query_time, |
| }) |
| assert started.status_code == 202, started.text |
| run_id = started.json()["id"] |
| job = settings.data_dir / "runs" / run_id |
| deadline = time.monotonic() + 60 |
| while time.monotonic() < deadline: |
| current = client.get(f"/api/runs/{run_id}") |
| assert current.status_code == 200 |
| if current.json()["status"] in {"completed", "failed", "cancelled"}: |
| break |
| time.sleep(0.05) |
| else: |
| pytest.fail("The real demo worker did not finish within 60 seconds.") |
| result = SimpleNamespace( |
| returncode=0 if current.json()["status"] == "completed" else 1, |
| stdout=(job / "worker.log").read_text(), stderr="", |
| ) |
| else: |
| result = subprocess.run([sys.executable, "-m", "demo.runner", "--job-dir", str(job)], |
| cwd=ROOT, env=env, text=True, capture_output=True, timeout=60) |
| finally: |
| server.shutdown() |
| server.server_close() |
| status = json.loads((job / "status.json").read_text()) |
| assert result.returncode == 0, (status, result.stdout[-1000:], result.stderr[-1000:]) |
| assert status["status"] == "completed" |
| assert status["answer"] == "Red." |
| assert status["queryTime"] == actual_time |
| assert status["plan"]["strategy"] == "replay 0 seconds then answer" |
| assert status["observations"] and status["memory"]["working"] |
| assert any(event["name"] == "replay" for event in status["events"]) |
| assert replay_requested and replay_received |
| historical = Image.open(io.BytesIO(base64.b64decode(replay_received[0].split(",", 1)[1]))) |
| red, green, blue = historical.getpixel((20, 20)) |
| assert red > 200 and green < 30 and blue < 30 |
| trajectory = json.loads((job / "trajectory.json").read_text()) |
| assert trajectory["question"] == "What color appeared at the beginning?" |
| steps = [json.loads(line) for line in (job / "steps.jsonl").read_text().splitlines()] |
| assert any(step["parsed_tool_calls"] for step in steps) |
| assert not any("*** META-QUESTION" in json.dumps(item) for item in seen) |
| assert status["modelCalls"] == len(seen) |
|
|
|
|
| def test_worker_diagnostics_redact_credentials(monkeypatch, capsys): |
| from demo.runner import diagnostic_error |
|
|
| monkeypatch.setenv("DEMO_API_KEY", "private-test-key") |
| try: |
| raise AttributeError("Provider failed for private-test-key and Bearer another-secret") |
| except AttributeError as exc: |
| diagnostic_error(exc) |
| output = capsys.readouterr().err |
| assert "AttributeError" in output and "test_runner.py" in output |
| assert "private-test-key" not in output and "another-secret" not in output |
|
|
