from __future__ import annotations import json import os from pathlib import Path import pytest from PIL import Image from recoil.core.paths import ProjectPaths from recoil.pipeline._lib.sidecar import compute_sha256 from recoil.pipeline.tools import gen_sublocations LOCATION = "int_lower_decks_maintenance_shaft" def _write_noise_png(path: Path, size: tuple[int, int] = (64, 64)) -> None: path.parent.mkdir(parents=True, exist_ok=True) w, h = size img = Image.frombytes("RGB", (w, h), os.urandom(w * h * 3)) img.save(path, "PNG") def _write_json(path: Path, payload: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2), encoding="utf-8") def _make_project( tmp_path: Path, sublocations: dict | None = None, *, sheets: tuple[int, ...] = (1,), ) -> ProjectPaths: root = tmp_path / "tartarus" bible_location: dict = {"description": "A lower-decks vertical shaft."} if sublocations is not None: bible_location["sublocations"] = sublocations _write_json( root / "_pipeline" / "state" / "visual" / "global_bible.json", {"locations": {LOCATION: bible_location}}, ) for version in sheets: _write_noise_png(root / "assets" / "loc" / LOCATION / "sheets" / f"sheet_v{version}.png") return ProjectPaths.from_root(root) def _patch_paths(monkeypatch: pytest.MonkeyPatch, paths: ProjectPaths) -> None: monkeypatch.setattr( gen_sublocations.ProjectPaths, "for_project", classmethod(lambda cls, project: paths), ) def _patch_adjacency(monkeypatch: pytest.MonkeyPatch, pairs: list[list[str]] | None = None) -> None: monkeypatch.setattr( gen_sublocations, "_propose_adjacency_pairs", lambda *_args, **_kwargs: pairs or [], ) def _fake_dispatch_factory(tmp_path: Path, calls: list[dict]): def fake_dispatch(project, location_id, name, prompt, quality, reference_images): calls.append( { "project": project, "location_id": location_id, "name": name, "prompt": prompt, "quality": quality, "reference_images": [Path(p) for p in reference_images], } ) out = tmp_path / f"raw_{len(calls)}.png" _write_noise_png(out) return out return fake_dispatch def _registry(paths: ProjectPaths) -> dict: path = paths.asset_look_dir("loc", LOCATION, "base") / "location.json" return json.loads(path.read_text(encoding="utf-8")) def test_missing_bible_block_errors(tmp_path, monkeypatch): paths = _make_project(tmp_path, None) _patch_paths(monkeypatch, paths) with pytest.raises(gen_sublocations.GenSublocationsError) as exc: gen_sublocations.generate_sublocations("tartarus", LOCATION, dry_run=True) assert "location has no bible sublocations block" in str(exc.value) def test_probe_fires_one_ref_and_writes_no_registry(tmp_path, monkeypatch, capsys): paths = _make_project( tmp_path, { "shaft_lip": {"description": "The upper lip of the shaft."}, "pod_platform": {"description": "A small platform around the pod."}, }, ) _patch_paths(monkeypatch, paths) calls: list[dict] = [] monkeypatch.setattr( gen_sublocations, "_dispatch_sublocation_ref", _fake_dispatch_factory(tmp_path, calls), ) result = gen_sublocations.generate_sublocations("tartarus", LOCATION, probe=True) assert result["registry_written"] is False assert [call["name"] for call in calls] == ["shaft_lip"] base = paths.asset_look_dir("loc", LOCATION, "base") ref = base / "sublocations" / "sublocation_shaft_lip_v01.png" assert ref.is_file() assert (ref.parent / (ref.name + ".json")).is_file() assert not (base / "location.json").exists() out = capsys.readouterr().out assert "tariff-estimated" in out assert "verify actual spend on the fal dashboard before the full run" in out def test_version_pick_uses_highest_master_plate(tmp_path, monkeypatch): paths = _make_project( tmp_path, {"shaft_lip": {"description": "The upper lip of the shaft."}}, sheets=(1, 3), ) _patch_paths(monkeypatch, paths) calls: list[dict] = [] monkeypatch.setattr( gen_sublocations, "_dispatch_sublocation_ref", _fake_dispatch_factory(tmp_path, calls), ) gen_sublocations.generate_sublocations("tartarus", LOCATION, probe=True) sheet_v3 = paths.get_location_sheets_dir(LOCATION) / "sheet_v3.png" assert calls[0]["reference_images"] == [sheet_v3] ref = paths.asset_look_dir("loc", LOCATION, "base") / "sublocations" / "sublocation_shaft_lip_v01.png" sidecar = json.loads((ref.parent / (ref.name + ".json")).read_text(encoding="utf-8")) assert sidecar["source_asset"] == "sheets/sheet_v3.png" assert sidecar["source_sha256"] == compute_sha256(sheet_v3) def test_landing_never_overwrites_existing_v01(tmp_path, monkeypatch): paths = _make_project( tmp_path, {"shaft_lip": {"description": "The upper lip of the shaft."}}, ) _patch_paths(monkeypatch, paths) _patch_adjacency(monkeypatch) calls: list[dict] = [] monkeypatch.setattr( gen_sublocations, "_dispatch_sublocation_ref", _fake_dispatch_factory(tmp_path, calls), ) ref = paths.asset_look_dir("loc", LOCATION, "base") / "sublocations" / "sublocation_shaft_lip_v01.png" _write_noise_png(ref) before = compute_sha256(ref) gen_sublocations.generate_sublocations("tartarus", LOCATION) assert calls == [] assert compute_sha256(ref) == before assert not (ref.parent / "sublocation_shaft_lip_v02.png").exists() def test_failed_validate_ref_file_gets_no_registry_entry(tmp_path, monkeypatch): paths = _make_project( tmp_path, {"shaft_lip": {"description": "The upper lip of the shaft."}}, ) _patch_paths(monkeypatch, paths) _patch_adjacency(monkeypatch) monkeypatch.setattr( gen_sublocations, "_dispatch_sublocation_ref", _fake_dispatch_factory(tmp_path, []), ) def fail_sublocation_refs(path: Path) -> str | None: if path.name.startswith("sublocation_"): return "forced validation failure" return None monkeypatch.setattr(gen_sublocations, "validate_ref_file", fail_sublocation_refs) result = gen_sublocations.generate_sublocations("tartarus", LOCATION) assert result["failed"] == [ {"sublocation": "shaft_lip", "error": "forced validation failure"} ] # New contract: a fresh location whose every generation failed stays # UNDECOMPOSED — creating an empty registry would flip consumers from # everything-permitted to nothing-allowed (fail-closed regression). assert result["registry_written"] is False base = paths.asset_look_dir("loc", LOCATION, "base") assert not (base / "location.json").exists() ref = base / "sublocations" / "sublocation_shaft_lip_v01.png" assert not (ref.parent / (ref.name + ".json")).exists() def test_registry_source_sha_is_bible_description_hash_not_plate_hash(tmp_path, monkeypatch): desc = "The upper lip of the shaft." paths = _make_project(tmp_path, {"shaft_lip": {"description": desc}}) _patch_paths(monkeypatch, paths) _patch_adjacency(monkeypatch) monkeypatch.setattr( gen_sublocations, "_dispatch_sublocation_ref", _fake_dispatch_factory(tmp_path, []), ) gen_sublocations.generate_sublocations("tartarus", LOCATION) entry = _registry(paths)["sublocations"]["shaft_lip"] assert entry["source_sha256"] == gen_sublocations.bible_desc_sha256(desc) assert entry["source_sha256"] != compute_sha256(paths.get_location_sheets_dir(LOCATION) / "sheet_v1.png") def test_restamp_converts_plate_hash_preserving_refs_and_adjacency(tmp_path, monkeypatch): desc = "The upper lip of the shaft." paths = _make_project(tmp_path, {"shaft_lip": {"description": desc}}) _patch_paths(monkeypatch, paths) ref = paths.asset_look_dir("loc", LOCATION, "base") / "sublocations" / "sublocation_shaft_lip_v01.png" _write_noise_png(ref) plate_hash = compute_sha256(paths.get_location_sheets_dir(LOCATION) / "sheet_v1.png") _write_json( paths.asset_look_dir("loc", LOCATION, "base") / "location.json", { "schema_version": 1, "location_id": LOCATION, "sublocations": { "shaft_lip": { "ref": "sublocations/sublocation_shaft_lip_v01.png", "source_sha256": plate_hash, } }, "adjacency": [["shaft_lip", "pod_platform"]], }, ) gen_sublocations.generate_sublocations("tartarus", LOCATION, restamp=True) registry = _registry(paths) assert registry["sublocations"]["shaft_lip"]["ref"] == "sublocations/sublocation_shaft_lip_v01.png" assert registry["sublocations"]["shaft_lip"]["source_sha256"] == gen_sublocations.bible_desc_sha256(desc) assert registry["adjacency"] == [["shaft_lip", "pod_platform"]] def test_registry_key_subset_violation_errors(tmp_path, monkeypatch): paths = _make_project(tmp_path, {"shaft_lip": {"description": "The upper lip."}}) _patch_paths(monkeypatch, paths) _write_json( paths.asset_look_dir("loc", LOCATION, "base") / "location.json", { "schema_version": 1, "location_id": LOCATION, "sublocations": { "unknown": { "ref": "sublocations/sublocation_unknown_v01.png", "source_sha256": "old", } }, "adjacency": [], }, ) with pytest.raises(gen_sublocations.GenSublocationsError) as exc: gen_sublocations.generate_sublocations("tartarus", LOCATION, restamp=True) assert "unknown" in str(exc.value) def test_adjacency_unconfirmed_writes_empty_adjacency_and_proposal(tmp_path, monkeypatch): paths = _make_project( tmp_path, { "shaft_lip": {"description": "The upper lip."}, "pod_platform": {"description": "The lower platform."}, }, ) _patch_paths(monkeypatch, paths) _patch_adjacency(monkeypatch, [["shaft_lip", "pod_platform"]]) base = paths.asset_look_dir("loc", LOCATION, "base") _write_noise_png(base / "sublocations" / "sublocation_shaft_lip_v01.png") _write_noise_png(base / "sublocations" / "sublocation_pod_platform_v01.png") gen_sublocations.generate_sublocations("tartarus", LOCATION) registry = _registry(paths) assert registry["adjacency"] == [] proposal = json.loads((base / "adjacency_proposal.json").read_text(encoding="utf-8")) assert proposal["adjacency"] == [["shaft_lip", "pod_platform"]] def test_adjacency_confirm_writes_pairs(tmp_path, monkeypatch): paths = _make_project( tmp_path, { "shaft_lip": {"description": "The upper lip."}, "pod_platform": {"description": "The lower platform."}, }, ) _patch_paths(monkeypatch, paths) _patch_adjacency(monkeypatch, [["shaft_lip", "pod_platform"]]) base = paths.asset_look_dir("loc", LOCATION, "base") _write_noise_png(base / "sublocations" / "sublocation_shaft_lip_v01.png") _write_noise_png(base / "sublocations" / "sublocation_pod_platform_v01.png") gen_sublocations.generate_sublocations( "tartarus", LOCATION, adjacency_confirm=True, ) assert _registry(paths)["adjacency"] == [["shaft_lip", "pod_platform"]] assert not (base / "adjacency_proposal.json").exists()