"""Audit-F4 acceptance: coverage_planner resolves previz + heroes from v3 paths.""" from PIL import Image from recoil.pipeline._lib.derivation_sha import shotset_hash from recoil.pipeline.orchestrator.coverage_planner import ( CoveragePass, CoverageSegment, resolve_blueprint, resolve_previz_frame, ) def _mk_project(tmp_path, monkeypatch): proj_root = tmp_path / "projects" / "tartarus" proj_root.mkdir(parents=True) monkeypatch.setattr( "recoil.core.paths.projects_root", lambda: tmp_path / "projects" ) return proj_root def _write_valid_jpeg(path): image = Image.new("RGB", (512, 512)) pixels = image.load() for y in range(512): for x in range(512): pixels[x, y] = ((x * 3) % 256, (y * 5) % 256, (x + y) % 256) image.save(path, format="JPEG", quality=95) assert path.stat().st_size >= 1024 def test_previz_resolves_from_v3_prep_dir(tmp_path, monkeypatch): proj = _mk_project(tmp_path, monkeypatch) prep = proj / "prep" / "ep_001" prep.mkdir(parents=True) (prep / "shot_005.png").write_bytes(b"\x89PNG") got = resolve_previz_frame("tartarus", 1, "EP001_SH05") assert got is not None and got.endswith("shot_005.png") assert "/prep/ep_001/" in got and "/output/" not in got def test_previz_resolves_numbered_take_from_v3_prep_dir(tmp_path, monkeypatch): proj = _mk_project(tmp_path, monkeypatch) prep = proj / "prep" / "ep_001" prep.mkdir(parents=True) (prep / "shot_005_take001.png").write_bytes(b"\x89PNG") got = resolve_previz_frame("tartarus", 1, "EP001_SH05") assert got is not None and got.endswith("shot_005_take001.png") assert "/prep/ep_001/" in got and "/output/" not in got def test_previz_ignores_v1_output_previs(tmp_path, monkeypatch): proj = _mk_project(tmp_path, monkeypatch) legacy = proj / "output" / "previs" / "ep_001" legacy.mkdir(parents=True) (legacy / "shot_005_take001.png").write_bytes(b"\x89PNG") assert resolve_previz_frame("tartarus", 1, "EP001_SH05") is None def test_sh05_sh05a_no_collision(tmp_path, monkeypatch): proj = _mk_project(tmp_path, monkeypatch) prep = proj / "prep" / "ep_001" prep.mkdir(parents=True) (prep / "shot_005.png").write_bytes(b"\x89PNG") (prep / "shot_005a.png").write_bytes(b"\x89PNG") (prep / "shot_005a_take002.png").write_bytes(b"\x89PNG") sh05 = resolve_previz_frame("tartarus", 1, "EP001_SH05") sh05a = resolve_previz_frame("tartarus", 1, "EP001_SH05A") assert sh05 is not None and sh05.endswith("shot_005.png") assert "shot_005a" not in sh05 assert sh05a is not None and sh05a.endswith("shot_005a_take002.png") # newest take wins def test_blueprint_hero_resolves_from_v3_hero_file(tmp_path, monkeypatch): proj = _mk_project(tmp_path, monkeypatch) base = proj / "assets" / "char" / "jade" / "base" base.mkdir(parents=True) _write_valid_jpeg(base / "jade_identity.jpeg") shots = [ { "shot_id": "EP001_SH01", "asset_data": {"characters": [{"char_id": "JADE"}]}, } ] path, source = resolve_blueprint(shots, "tartarus", 1) # source must be "hero" and path must be the v3 hero (exact assertion # tolerant to resolve_hero's extension probing) assert source == "hero" assert "/assets/char/jade/base/" in (path or "") def test_newer_take_preferred_over_base(tmp_path, monkeypatch): """Merge-gate finding 2026-06-10: base frame must NOT shadow newer takes.""" proj = _mk_project(tmp_path, monkeypatch) prep = proj / "prep" / "ep_001" prep.mkdir(parents=True) (prep / "shot_005.png").write_bytes(b"\x89PNG") (prep / "shot_005_take001.png").write_bytes(b"\x89PNG") (prep / "shot_005_take002.png").write_bytes(b"\x89PNG") got = resolve_previz_frame("tartarus", 1, "EP001_SH05") assert got is not None and got.endswith("shot_005_take002.png") def test_take_ordering_is_numeric_not_lexical(tmp_path, monkeypatch): """Merge-gate round 3: _take10 must beat _take9 (lexical sort gets this wrong).""" proj = _mk_project(tmp_path, monkeypatch) prep = proj / "prep" / "ep_001" prep.mkdir(parents=True) (prep / "shot_005_take9.png").write_bytes(b"\x89PNG") (prep / "shot_005_take10.png").write_bytes(b"\x89PNG") got = resolve_previz_frame("tartarus", 1, "EP001_SH05") assert got is not None and got.endswith("shot_005_take10.png") def _coverage_pass_with_segments( source_shot_ids: list[str], *, shot_range: tuple[str, str] = ("EP001_SH01", "EP001_SH03"), ) -> CoveragePass: return CoveragePass( pass_id="EP001_PASS_001", episode_id="ep_001", shot_range=shot_range, camera_side="A", label="coverage", focus_character="JADE", pass_type="character", segments=[ CoverageSegment( segment_index=index, source_shot_id=source_shot_id, shot_type="MS", duration_s=5, prompt=f"Segment {index}", is_wildcard=source_shot_id == "wildcard", ) for index, source_shot_id in enumerate(source_shot_ids, start=1) ], ) def test_coverage_pass_to_dict_shotset_hash_is_order_independent(): coverage_pass = _coverage_pass_with_segments( ["EP001_SH03", "EP001_SH01", "EP001_SH02"] ) assert coverage_pass.to_dict()["shotset_hash"] == shotset_hash( ["EP001_SH01", "EP001_SH02", "EP001_SH03"] ) def test_coverage_pass_to_dict_shotset_hash_excludes_wildcard_segment(): coverage_pass = _coverage_pass_with_segments( ["EP001_SH03", "wildcard", "EP001_SH02"] ) assert coverage_pass.to_dict()["shotset_hash"] == shotset_hash( ["EP001_SH02", "EP001_SH03"] ) def test_coverage_pass_to_dict_shotset_hash_uses_shot_range_for_wildcard_only(): coverage_pass = _coverage_pass_with_segments( ["wildcard"], shot_range=("EP001_SH04", "EP001_SH05"), ) assert coverage_pass.to_dict()["shotset_hash"] == shotset_hash( ["EP001_SH04", "EP001_SH05"] )