from __future__ import annotations import json from pathlib import Path import pytest from recoil.execution.step_types import ProjectPaths from recoil.pipeline._lib import derivation_manifest from recoil.pipeline.cli import generate from recoil.pipeline.core.persistence import scene_path from recoil.pipeline.orchestrator import ingest_pipeline from recoil.pipeline.orchestrator.ingest_pipeline import ( IngestPipeline, LocationUnresolvedError, ) PROJECT = "fixture" @pytest.fixture(autouse=True) def _projects_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: root = tmp_path / "projects" root.mkdir() (root / ".recoil-data-root").touch() project_root = root / PROJECT project_root.mkdir() monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(root)) return project_root def _shot(batch_index: int, shot_index: int) -> dict: ordinal = (batch_index - 1) * 3 + shot_index return { "shot_id": f"EP001_SH{ordinal:02d}", "scene_index": batch_index, "pipeline": "video", "video_model": "seeddance-2.0", "asset_data": { "characters": [], "location_id": f"LOC_{batch_index}", }, "prompt_data": {"shot_type": "MS"}, "routing_data": { "target_editorial_duration_s": 2.0, "is_env_only": False, "has_dialogue": False, }, "aspect_ratio": "9:16", } def _plan_payload() -> dict: return { "episode_id": "ep_001", "project": PROJECT, "shots": [ _shot(batch_index, shot_index) for batch_index in range(1, 5) for shot_index in range(1, 4) ], } def _write_plan() -> Path: paths = ProjectPaths.for_episode(PROJECT, 1) paths.plans_dir.mkdir(parents=True, exist_ok=True) plan_path = paths.plans_dir / "ep_001_plan.json" plan_path.write_text(json.dumps(_plan_payload(), indent=2), encoding="utf-8") return plan_path def _set_rederive_argv(monkeypatch: pytest.MonkeyPatch, *extra: str) -> None: monkeypatch.setattr( generate.sys, "argv", [ "generate.py", "rederive", "--project", PROJECT, "--episode", "1", *extra, ], ) def test_rederive_runs_plan_pass_and_consumes_regenerated_plan( monkeypatch: pytest.MonkeyPatch, ) -> None: calls: list[int] = [] def fake_load_bible(self): # noqa: ANN001 return {"stub": "bible"} def fake_run_storyboard_pass(self, episode: int, bible): # noqa: ANN001 calls.append(episode) assert bible == {"stub": "bible"} plan_path = self.plans_dir / "ep_001_plan.json" plan_path.write_text( json.dumps(_plan_payload(), indent=2), encoding="utf-8", ) return {"sentinel": "plan"} monkeypatch.setattr(generate.IngestPipeline, "_load_bible", fake_load_bible) monkeypatch.setattr( generate.IngestPipeline, "run_storyboard_pass", fake_run_storyboard_pass, ) _set_rederive_argv( monkeypatch, "--skip-camera-test", "--skip-extract", "--dry-run", ) code = generate.main() assert code == 0 assert calls == [1] assert ( ProjectPaths.for_episode(PROJECT, 1).plans_dir / "ep_001_plan.json" ).exists() def test_rederive_skip_plan_bypasses_plan_pass_and_preserves_existing_plan( monkeypatch: pytest.MonkeyPatch, ) -> None: plan_path = _write_plan() before = plan_path.read_bytes() calls: list[int] = [] def fake_run_storyboard_pass(self, episode: int, bible): # noqa: ANN001 calls.append(episode) raise AssertionError("run_storyboard_pass must not run with --skip-plan") monkeypatch.setattr( generate.IngestPipeline, "run_storyboard_pass", fake_run_storyboard_pass, ) _set_rederive_argv( monkeypatch, "--skip-camera-test", "--skip-plan", "--skip-extract", "--dry-run", ) code = generate.main() assert code == 0 assert calls == [] assert plan_path.read_bytes() == before def test_rederive_plan_pass_validation_raise_aborts_before_stage_c( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: def fake_load_bible(self): # noqa: ANN001 return {"stub": "bible"} def fake_run_storyboard_pass(self, episode: int, bible): # noqa: ANN001 raise ValueError("motion_line ...") monkeypatch.setattr(generate.IngestPipeline, "_load_bible", fake_load_bible) monkeypatch.setattr( generate.IngestPipeline, "run_storyboard_pass", fake_run_storyboard_pass, ) _set_rederive_argv( monkeypatch, "--skip-camera-test", "--skip-extract", "--dry-run", ) code = generate.main() captured = capsys.readouterr() assert code == 1 assert "Plan Pass failed validation" in captured.err assert "motion_line ..." in captured.err assert not scene_path(PROJECT, "ep_001", "BATCH_001").exists() def test_rederive_stamps_location_unresolved_on_abort( monkeypatch: pytest.MonkeyPatch, ) -> None: def fake_load_bible(self): # noqa: ANN001 return {"stub": "bible"} def fake_run_storyboard_pass(self, *args, **kwargs): # noqa: ANN001 raise LocationUnresolvedError("unresolved", shot_index=2, hint="INT. NOWHERE") monkeypatch.setattr(IngestPipeline, "_load_bible", fake_load_bible) monkeypatch.setattr(IngestPipeline, "run_storyboard_pass", fake_run_storyboard_pass) rc = generate._run_rederive_cli( [ "--project", PROJECT, "--episode", "1", "--skip-camera-test", "--skip-extract", "--dry-run", ] ) assert rc == generate.EXIT_PARTIAL manifest = derivation_manifest.load(PROJECT, 1) assert manifest["health"]["flags"]["location.unresolved"]["shot_index"] == 2 assert manifest["health"]["flags"]["location.unresolved"]["hint"] == "INT. NOWHERE" def test_rederive_plan_pass_runtime_contract_does_not_require_api_keys( monkeypatch: pytest.MonkeyPatch, ) -> None: assert ingest_pipeline.CAMERA_TEST_MODEL.startswith("claude-") assert ingest_pipeline.STORYBOARD_MODEL.startswith("claude-") monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("GEMINI_API_KEY", raising=False) calls: list[int] = [] def fake_load_bible(self): # noqa: ANN001 return {"stub": "bible"} def fake_run_storyboard_pass(self, episode: int, bible): # noqa: ANN001 calls.append(episode) (self.plans_dir / "ep_001_plan.json").write_text( json.dumps(_plan_payload(), indent=2), encoding="utf-8", ) return {"sentinel": "plan"} monkeypatch.setattr(generate.IngestPipeline, "_load_bible", fake_load_bible) monkeypatch.setattr( generate.IngestPipeline, "run_storyboard_pass", fake_run_storyboard_pass, ) _set_rederive_argv( monkeypatch, "--skip-camera-test", "--skip-extract", "--dry-run", ) code = generate.main() assert code == 0 assert calls == [1]