"""Boundary tests for engine entity routes (Phase 16 — Law 11). Each test exercises ONE behavioral invariant of the HTTP contract. Adapter unit tests live alongside the adapter modules; these are pure boundary tests through FastAPI's TestClient. """ from __future__ import annotations import json from pathlib import Path import pytest from fastapi.testclient import TestClient # ── Test fixtures: synthesize a tiny projects/ tree + receipts log ────────── @pytest.fixture(scope="session", autouse=True) def _tmp_db(tmp_path_factory: pytest.TempPathFactory) -> Path: """Redirect the workspace_state DB so the import doesn't touch ~/.recoil.""" tmp = tmp_path_factory.mktemp("recoil_engine_db") / "test.db" import recoil.api.db as db_mod db_mod.DB_PATH = tmp db_mod.init_db() return tmp @pytest.fixture(scope="session") def projects_root(tmp_path_factory: pytest.TempPathFactory) -> Path: root = tmp_path_factory.mktemp("recoil_projects") (root / ".recoil-data-root").write_text("recoil-data-root\n") # tartarus — loads cleanly tart = root / "tartarus" / "_pipeline" / "state" / "visual" (tart / "shots").mkdir(parents=True) (tart / "global_bible.json").write_text( json.dumps({"project": "Tartarus", "total_episodes": 3, "aspect": "9_16"}), encoding="utf-8", ) # Two beats for tartarus EP001 (tart / "shots" / "EP001_SH01.json").write_text( json.dumps( { "shot_id": "EP001_SH01", "episode_id": "EP001", "pipeline": "previz", "model": "gemini-3.1-flash-image-preview", "status": "pending_qc", "takes": [ { "take_id": "EP001_SH01_T001", "take_number": 1, "file_path": "output/previs/ep_001/shot_001_take1.png", "compiled_prompt": "Close-up, 50mm. Jade salvages copper.", "cost_usd": 0.039, "model": "gemini-3.1-flash-image-preview", "timestamp": 1773711691.6, "rejected": False, "gate_1": { "passed": True, "details": {"anatomy": {"pass": True, "reason": "ok"}}, }, }, { "take_id": "EP001_SH01_T002", "take_number": 2, "file_path": "output/previs/ep_001/shot_001_take2.png", "cost_usd": 0.039, "timestamp": 1773712223.8, "rejected": True, }, ], } ), encoding="utf-8", ) (tart / "shots" / "EP001_SH02.json").write_text( json.dumps( { "shot_id": "EP001_SH02", "episode_id": "EP001", "model": "kling-v2-master", "status": "complete", "takes": [ { "take_id": "EP001_SH02_T001", "take_number": 1, "file_path": "output/video/ep_001/sh02_take1.mp4", "cost_usd": 0.45, "model": "kling-v2-master", "timestamp": 1773720000, }, ], } ), encoding="utf-8", ) # legacy-only — no global_bible.json (root / "legacy-show" / "state" / "visual").mkdir(parents=True) # _archive — should be skipped by reserved-slug filter (root / "_archive").mkdir(parents=True) return root @pytest.fixture(scope="session") def receipts_log(tmp_path_factory: pytest.TempPathFactory) -> Path: log_dir = tmp_path_factory.mktemp("recoil_receipts") p = log_dir / "receipts.jsonl" rows = [ { "receipt_id": "rcpt_t1", "modality": "image_t2i", "caller_id": "run_shot", "project": "tartarus", "episode": 1, "shot_id": "EP001_SH01", "timestamp_utc": "2026-04-28T05:34:37Z", "run_result": { "id": "x", "modality": "image_t2i", "metadata": {"final_state": "succeeded", "cost_usd": 0.039}, "success": True, "error": None, }, "provenance": {"model": "gemini-3.1-flash-image-preview"}, }, { "receipt_id": "rcpt_t2", "modality": "image_t2i", "caller_id": "run_shot", "project": "tartarus", "episode": 1, "shot_id": "EP001_SH02", "timestamp_utc": "2026-04-28T05:35:00Z", "run_result": { "id": "y", "modality": "image_t2i", "metadata": {"final_state": "failed", "cost_usd": 0.0}, "success": False, "error": "RuntimeError: Infra failure", }, "provenance": {"model": "test-model"}, }, ] p.write_text("\n".join(json.dumps(r) for r in rows) + "\n", encoding="utf-8") return p @pytest.fixture(scope="session") def memory_dir(tmp_path_factory: pytest.TempPathFactory) -> Path: d = tmp_path_factory.mktemp("recoil_memory") (d / "LEARNINGS.md").write_text( """# Engine Learnings ## Entries ### 2026-03-25 [script] - **ID:** L001 - **Learning:** Spatial logic breaks are the dominant failure mode. - **Evidence:** brief shows 11/60 spatial findings. - **Implication:** Avoid CONTINUOUS across level changes. - **Status:** provisional """, encoding="utf-8", ) (d / "ANTI_PATTERNS.md").write_text( """# Engine Anti-Patterns ## Script Anti-Patterns ### Off-Screen Threat Resolution at Cliffhangers - **ID:** A001 - **Status:** provisional - **What it is:** Episode N+1 resolves the cliffhanger via exposition. - **Why it fails:** Breaks the cliffhanger contract. - **Instead, do:** Open with a physical action. ## Visual Anti-Patterns ## Retired """, encoding="utf-8", ) return d @pytest.fixture(autouse=True) def _bind_env( monkeypatch: pytest.MonkeyPatch, projects_root: Path, receipts_log: Path, memory_dir: Path, ): """Per-test env binding via monkeypatch — automatic cleanup, no session leak. Previously this fixture was session-scoped and used os.environ directly, which leaked RECOIL_PROJECTS_ROOT across the whole pytest session and caused unrelated tests (e.g. test_ttyd_lifecycle) to look up projects against the synthetic test root. monkeypatch.setenv restores the prior value at function teardown. """ monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(projects_root)) monkeypatch.setenv("RECOIL_RECEIPTS_LOG", str(receipts_log)) monkeypatch.setenv("RECOIL_ENGINE_MEMORY_DIR", str(memory_dir)) @pytest.fixture def client() -> TestClient: from recoil.api.main import app return TestClient(app) # ── /api/projects ────────────────────────────────────────────────────────── def test_projects_returns_real_project(client: TestClient) -> None: r = client.get("/api/projects") assert r.status_code == 200 items = r.json() assert any(p["id"] == "tartarus" for p in items) for p in items: assert p["aspect"] in ("9_16", "16_9", "1_1", "4_3") assert p["schemaVersion"] == 1 def test_legacy_project_excluded_from_list(client: TestClient) -> None: r = client.get("/api/projects") assert r.status_code == 200 ids = [p["id"] for p in r.json()] assert "legacy-show" not in ids assert "_archive" not in ids def test_get_project_by_id(client: TestClient) -> None: r = client.get("/api/projects/tartarus") assert r.status_code == 200 p = r.json() assert p["id"] == "tartarus" assert p["name"] == "Tartarus" assert p["aspect"] == "9_16" def test_unknown_project_404(client: TestClient) -> None: r = client.get("/api/projects/does-not-exist") assert r.status_code == 404 def test_legacy_project_get_returns_410(client: TestClient) -> None: r = client.get("/api/projects/legacy-show") assert r.status_code == 410 assert "legacy" in r.json()["detail"].lower() # ── /api/projects/.../beats ─────────────────────────────────────────────── def test_list_beats_for_episode(client: TestClient) -> None: """P3: scene_id is now the synthetic name `__synthetic_scene_1`.""" r = client.get( "/api/projects/tartarus/episodes/EP001/scenes/EP001__synthetic_scene_1/beats" ) assert r.status_code == 200 beats = r.json() ids = sorted(b["id"] for b in beats) assert ids == ["EP001_SH01", "EP001_SH02"] for b in beats: assert b["schemaVersion"] == 1 assert b["status"] in ("pending", "running", "blocked", "locked", "draft") assert b["takes"] >= 0 def test_list_beats_unknown_episode_returns_404(client: TestClient) -> None: """P3: bogus episode_id → 404 (was 200/[]; new behavior is fail-loud).""" r = client.get( "/api/projects/tartarus/episodes/EP999/scenes/EP999__synthetic_scene_1/beats" ) assert r.status_code == 404 def test_list_beats_bogus_scene_id_returns_404(client: TestClient) -> None: """P3: bogus scene_id (not the synthetic name) → 404.""" r = client.get("/api/projects/tartarus/episodes/EP001/scenes/sc01/beats") assert r.status_code == 404 # ── P3: /api/projects/{pid}/episodes ────────────────────────────────────── def test_get_episodes_returns_synthesized_list_for_tartarus( client: TestClient, ) -> None: r = client.get("/api/projects/tartarus/episodes") assert r.status_code == 200 eps = r.json() assert len(eps) >= 1 assert all(ep["synthesized"] is True for ep in eps) # EP001 must appear (the fixture has two shots under it). assert any(ep["id"] == "EP001" for ep in eps) def test_get_episodes_unknown_project_returns_empty(client: TestClient) -> None: """Unknown project → empty episode list (no shots ⇒ no episodes).""" r = client.get("/api/projects/no-such-project/episodes") # validate_project_id passes for slug-shaped names; the projects_root # walk just yields no shot files, so the synthesized list is empty. assert r.status_code == 200 assert r.json() == [] def test_get_episodes_rejects_malformed_project_id(client: TestClient) -> None: r = client.get("/api/projects/UPPERCASE/episodes") assert r.status_code == 400 # ── P3: /api/projects/{pid}/episodes/{eid}/scenes ───────────────────────── def test_get_scenes_returns_one_synthetic_scene_per_episode( client: TestClient, ) -> None: r = client.get("/api/projects/tartarus/episodes/EP001/scenes") assert r.status_code == 200 scenes = r.json() assert len(scenes) == 1 assert scenes[0]["id"] == "EP001__synthetic_scene_1" assert scenes[0]["synthesized"] is True def test_get_scenes_for_unknown_episode_returns_404(client: TestClient) -> None: r = client.get("/api/projects/tartarus/episodes/EP999/scenes") assert r.status_code == 404 def test_get_beats_for_synthetic_scene_returns_real_beats( client: TestClient, ) -> None: r = client.get( "/api/projects/tartarus/episodes/EP001/scenes/EP001__synthetic_scene_1/beats" ) assert r.status_code == 200 beats = r.json() assert len(beats) > 0 def test_get_beats_for_bogus_scene_returns_404(client: TestClient) -> None: r = client.get("/api/projects/tartarus/episodes/EP001/scenes/MADE_UP_SCENE/beats") assert r.status_code == 404 def _make_missing_canonical_root(tmp_path: Path) -> Path: """Build a microdrama project whose single shot has NO resolvable episode_id (no `episode_id` field, no /ep_NNN/ output_path, no `_SH` delimiter in the filename, and not a REF_ asset). After Build A Phase 4 retired the filename-prefix fallback, this is a structurally-absent canonical field — the adapter must raise MissingCanonicalFieldError, which the route boundary surfaces as a 422.""" root = tmp_path / "mc_root" root.mkdir(parents=True, exist_ok=True) (root / ".recoil-data-root").write_text("recoil-data-root\n") vis = root / "missingcanon" / "_pipeline" / "state" / "visual" (vis / "shots").mkdir(parents=True) (vis / "global_bible.json").write_text( json.dumps({"project": "MissingCanon", "total_episodes": 1, "aspect": "9_16"}), encoding="utf-8", ) # Non-REF shot, no episode_id, no output_path, no _SH delimiter → unresolvable. (vis / "shots" / "REGEN_orphan.json").write_text( json.dumps({"shot_id": "REGEN_orphan", "status": "pending", "takes": []}), encoding="utf-8", ) return root def test_get_project_missing_episode_id_returns_422( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """B6 (was test_episodes_synthesis_fires_fallback): episode synthesis is now the canonical path, not a sanctioned fallback. When a non-REF shot has no resolvable episode_id, the per-project route returns 422 with the MissingCanonicalField body instead of laundering the absence.""" from recoil.api.main import app monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(_make_missing_canonical_root(tmp_path))) with TestClient(app) as c: r = c.get("/api/projects/missingcanon") assert r.status_code == 422 detail = r.json()["detail"] assert detail["error"] == "missing_canonical_field" assert detail["field"] == "episode_id" assert detail["project_id"] == "missingcanon" assert detail["fix_cli"] # non-empty remediation command def test_get_beats_missing_episode_id_returns_422( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """B6 (was test_scenes_synthesis_fires_fallback): the scenes/one-per-episode synthesis is canonical, not a fallback. A non-REF shot with an unresolvable episode_id surfaces as a 422 at the beats route (which calls list_beats), not a silent drop.""" from recoil.api.main import app monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(_make_missing_canonical_root(tmp_path))) with TestClient(app) as c: r = c.get( "/api/projects/missingcanon/episodes/EP001/scenes/EP001__synthetic_scene_1/beats" ) assert r.status_code == 422 detail = r.json()["detail"] assert detail["error"] == "missing_canonical_field" assert detail["field"] == "episode_id" assert detail["project_id"] == "missingcanon" # ── /api/beats/{id}/takes ───────────────────────────────────────────────── def test_list_takes_for_beat(client: TestClient) -> None: r = client.get("/api/beats/EP001_SH01/takes", params={"projectId": "tartarus"}) assert r.status_code == 200 takes = r.json() assert len(takes) == 2 assert takes[0]["beatId"] == "EP001_SH01" assert takes[0]["evalState"] == "pass" assert takes[1]["status"] == "failed" # rejected assert takes[0]["media"] == "still" def test_list_takes_unknown_beat_returns_empty(client: TestClient) -> None: r = client.get("/api/beats/EP999_SH01/takes", params={"projectId": "tartarus"}) assert r.status_code == 200 assert r.json() == [] # ── /api/beats/{id}/lineage ─────────────────────────────────────────────── def test_lineage_includes_prompt_step_output(client: TestClient) -> None: r = client.get("/api/beats/EP001_SH01/lineage", params={"projectId": "tartarus"}) assert r.status_code == 200 L = r.json() assert L["beatId"] == "EP001_SH01" kinds = {n["kind"] for n in L["nodes"]} assert "prompt" in kinds assert "step" in kinds or "sibling" in kinds assert "output" in kinds def test_lineage_unknown_beat_404(client: TestClient) -> None: r = client.get("/api/beats/EP999_SH99/lineage", params={"projectId": "tartarus"}) assert r.status_code == 404 # ── /api/events ─────────────────────────────────────────────────────────── def test_events_returns_receipts_plus_legacy(client: TestClient) -> None: r = client.get("/api/events") assert r.status_code == 200 events = r.json() ids = [e["id"] for e in events] assert "rcpt_t1" in ids assert "rcpt_t2" in ids # Legacy project also surfaces assert any(e["id"].startswith("legacy-project-") for e in events) def test_events_filter_by_severity_failure(client: TestClient) -> None: r = client.get("/api/events", params={"severities": ["failure"]}) assert r.status_code == 200 for e in r.json(): assert e["severity"] == "failure" def test_events_limit(client: TestClient) -> None: r = client.get("/api/events", params={"limit": 1}) assert r.status_code == 200 assert len(r.json()) <= 1 def test_events_scope_prefix(client: TestClient) -> None: r = client.get("/api/events", params={"scopePrefix": "tartarus"}) assert r.status_code == 200 for e in r.json(): assert e["scope"].startswith("tartarus") # ── /api/memory ─────────────────────────────────────────────────────────── def test_memory_returns_learning_and_anti_pattern(client: TestClient) -> None: r = client.get("/api/memory") assert r.status_code == 200 items = r.json() kinds = {m["kind"] for m in items} assert "learning" in kinds assert "anti-pattern" in kinds for m in items: assert m["schemaVersion"] == 1 assert isinstance(m["text"], str) # ── Schema shape parity with TS fixtures ────────────────────────────────── def test_pydantic_schema_shape_matches_fixtures(client: TestClient) -> None: """Catches Pydantic→TS drift early — fixture types and Pydantic shapes must agree on field names + nesting.""" r = client.get("/api/projects/tartarus") assert r.status_code == 200 p = r.json() for k in ("id", "name", "aspect", "score", "episodes", "schemaVersion"): assert k in p def test_workspace_state_routes_still_work(client: TestClient) -> None: """Phase 1 preservation: the workspace_state routes still mount under /api.""" r = client.get("/api/health") assert r.status_code == 200 assert r.json() == {"ok": True} # ── Debug R1 path-traversal hardening ───────────────────────────────────── # # Two attack shapes: # 1. ``..%2fevil`` is URL-decoded to ``../evil`` by Starlette → the path # no longer matches ``/api/projects/{project_id}`` → 404 from the # routing layer (NOT the handler). The traversal is structurally # impossible to reach the adapter, which is the strongest possible # block. # 2. Encoded ``%2e%2e`` ➜ ``..`` (or any non-encoded malformed ID like # ``Tartarus`` with uppercase) DOES reach the handler. The adapter # rejects with ValueError → 400. # # Both cases are tested. The contract: a malformed ID must NEVER produce # a 200 with leaked data. def test_get_project_routes_layer_blocks_dotdot_traversal( client: TestClient, ) -> None: """Encoded ``..%2f`` decodes to ``../`` and breaks routing → 404. The strongest possible block — the request can't even reach the adapter. """ r = client.get("/api/projects/..%2fevil%2fpath") assert r.status_code in (400, 404) # Critically, NOT 200 — the adapter must never serve content for this. assert r.status_code != 200 def test_get_project_rejects_uppercase_id(client: TestClient) -> None: """Single-segment malformed ID reaches handler → 400 from adapter.""" r = client.get("/api/projects/Tartarus") # uppercase rejected by regex assert r.status_code == 400 def test_get_project_rejects_double_dot_segment(client: TestClient) -> None: """A literal ``..`` segment (no slash) reaches the handler → 400.""" r = client.get("/api/projects/..") assert r.status_code in (400, 404, 405) assert r.status_code != 200 def test_list_takes_rejects_malformed_beat_id(client: TestClient) -> None: """Encoded slash in beat_id → routing-layer 404 (path no longer matches). Either 400 (adapter rejection) or 404 (route mismatch) is acceptable; a 200 with leaked data is the failure mode we're guarding against. """ r = client.get( "/api/beats/..%2fetc%2fpasswd/takes", params={"projectId": "tartarus"}, ) assert r.status_code in (400, 404) assert r.status_code != 200 def test_list_takes_rejects_malformed_project_id(client: TestClient) -> None: r = client.get( "/api/beats/EP001_SH01/takes", params={"projectId": "../evil"}, ) assert r.status_code == 400 def test_lineage_rejects_traversal_beat_id(client: TestClient) -> None: r = client.get( "/api/beats/..%2fetc/lineage", params={"projectId": "tartarus"}, ) assert r.status_code in (400, 404) assert r.status_code != 200 def test_list_beats_rejects_malformed_episode_id(client: TestClient) -> None: r = client.get("/api/projects/tartarus/episodes/..%2fevil/scenes/sc01/beats") assert r.status_code in (400, 404) assert r.status_code != 200 def test_list_beats_rejects_malformed_project_id_slug( client: TestClient, ) -> None: """Single-segment malformed project slug reaches handler → 400.""" r = client.get("/api/projects/UPPERCASE/episodes/EP001/scenes/sc01/beats") assert r.status_code == 400