#!/usr/bin/env python3 """Tests for workspace/sidecar.py — Universal Sidecar Module. Run: python3 -m pytest workspace/tests/test_sidecar.py -v """ import json import sys from pathlib import Path import pytest # ── Path setup ──────────────────────────────────────────────── _RECOIL_ROOT = Path(__file__).resolve().parent.parent.parent if str(_RECOIL_ROOT) not in sys.path: sys.path.insert(0, str(_RECOIL_ROOT)) from recoil.workspace.sidecar import ( # noqa: E402 SCHEMA_VERSION, SIDECAR_VALID_STATUSES, _sidecar_path, archive_with_sidecar, auto_stub_missing, create_stub_sidecar, ensure_sidecar, get_status, promote_to_canonical, read_sidecar, restore_from_archive, scan_for_missing_sidecars, set_status, write_sidecar, ) # R6 Phase 7 — write_pipeline_sidecar (_RETIRED) deleted; tests migrate to the SSOT. from recoil.pipeline._lib.sidecar import populate_sidecar, write_sidecar_dict # noqa: E402 # ── Fixtures ────────────────────────────────────────────────── @pytest.fixture def tmp_project(tmp_path): """Create a minimal project directory structure.""" project_dir = tmp_path / "test_project" output_dir = project_dir / "output" refs_dir = output_dir / "refs" / "characters" / "sadie" refs_dir.mkdir(parents=True) canonical_dir = output_dir / "refs" / "_canonical" / "characters" / "sadie" canonical_dir.mkdir(parents=True) (canonical_dir / "_meta").mkdir() archive_dir = output_dir / "_archive" archive_dir.mkdir() state_dir = project_dir / "state" / "visual" state_dir.mkdir(parents=True) return project_dir @pytest.fixture def sample_image(tmp_project): """Create a sample image file.""" img = tmp_project / "output" / "refs" / "characters" / "sadie" / "hero_v4.jpg" img.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100) # minimal JPEG header return img # ── Core Read/Write Tests ───────────────────────────────────── class TestSidecarPath: def test_sidecar_path_jpg(self, sample_image): sc = _sidecar_path(sample_image) assert sc.name == "hero_v4.jpg.json" assert sc.parent == sample_image.parent def test_sidecar_path_mp4(self, tmp_path): video = tmp_path / "scene_01.mp4" sc = _sidecar_path(video) assert sc.name == "scene_01.mp4.json" class TestReadWrite: def test_write_and_read(self, sample_image): data = {"status": "candidate", "notes": "test"} write_sidecar(sample_image, data) result = read_sidecar(sample_image) assert result is not None assert result["status"] == "candidate" assert result["notes"] == "test" assert result["schema_version"] == SCHEMA_VERSION assert "updated_at" in result def test_read_missing_returns_none(self, sample_image): assert read_sidecar(sample_image) is None def test_write_is_atomic(self, sample_image): """Verify atomic write creates the file and it's valid JSON.""" write_sidecar(sample_image, {"test": True}) sc = _sidecar_path(sample_image) assert sc.is_file() data = json.loads(sc.read_text()) assert data["test"] is True def test_write_overwrites(self, sample_image): write_sidecar(sample_image, {"version": 1}) write_sidecar(sample_image, {"version": 2}) result = read_sidecar(sample_image) assert result["version"] == 2 def test_read_corrupt_json_raises(self, sample_image): # Phase E.6 / Site #2 (workspace/sidecar.py:200): corrupt JSON now # raises SidecarCorruptError per Tenet 6 (was: silent return None). # See recoil/docs/silent-failure-inventory.md Site #2. from recoil.core.exceptions import SidecarCorruptError sc = _sidecar_path(sample_image) sc.write_text("not json {{{") with pytest.raises(SidecarCorruptError): read_sidecar(sample_image) class TestEnsureSidecar: def test_creates_stub_when_missing(self, sample_image): data = ensure_sidecar(sample_image) assert data["source"] == "manual_drop" assert data["status"] == "candidate" # Verify file was actually written assert _sidecar_path(sample_image).is_file() def test_returns_existing_when_present(self, sample_image): write_sidecar(sample_image, {"source": "pipeline", "status": "front-runner"}) data = ensure_sidecar(sample_image) assert data["source"] == "pipeline" assert data["status"] == "front-runner" class TestCreateStub: def test_stub_has_required_fields(self, sample_image): data = create_stub_sidecar(sample_image) assert data["schema_version"] == SCHEMA_VERSION assert data["source"] == "manual_drop" assert data["status"] == "candidate" assert "created_at" in data assert "updated_at" in data assert data["provenance"] == {} assert data["lineage"] == {} # ── Status Management Tests ─────────────────────────────────── class TestSetStatus: def test_set_valid_status(self, sample_image): for status in SIDECAR_VALID_STATUSES: data = set_status(sample_image, status) assert data["status"] == status def test_set_invalid_status_raises(self, sample_image): with pytest.raises(ValueError, match="Invalid status"): set_status(sample_image, "bogus") def test_extra_kwargs_merged(self, sample_image): data = set_status(sample_image, "pinned", notes="Best one yet") assert data["notes"] == "Best one yet" def test_creates_sidecar_if_missing(self, sample_image): data = set_status(sample_image, "pinned") assert data["source"] == "manual_drop" # auto-stubbed assert data["status"] == "pinned" class TestGetStatus: def test_returns_none_when_no_sidecar(self, sample_image): assert get_status(sample_image) is None def test_returns_correct_status(self, sample_image): set_status(sample_image, "candidate") assert get_status(sample_image) == "candidate" # ── Canonical Promotion Tests ───────────────────────────────── class TestPromoteToCanonical: def test_promotion_copies_file(self, sample_image, tmp_project): promote_to_canonical( sample_image, asset_type="characters", entity_id="sadie", project_dir=tmp_project, ) hero = ( tmp_project / "assets" / "char" / "sadie" / "base" / "sadie_identity.jpeg" ) assert hero.is_file() def test_promotion_writes_canonical_sidecar(self, sample_image, tmp_project): promote_to_canonical( sample_image, asset_type="characters", entity_id="sadie", project_dir=tmp_project, ) meta_sc = ( tmp_project / "assets" / "char" / "sadie" / "base" / "sadie_identity.jpeg.json" ) assert meta_sc.is_file() data = json.loads(meta_sc.read_text()) assert data["status"] == "canonical" def test_promotion_updates_casting_state(self, sample_image, tmp_project): promote_to_canonical( sample_image, asset_type="characters", entity_id="sadie", project_dir=tmp_project, ) casting_path = ( tmp_project / "_pipeline" / "state" / "visual" / "casting_state.json" ) assert casting_path.is_file() casting = json.loads(casting_path.read_text()) assert "characters" in casting assert "sadie" in casting["characters"] assert "hero_path" in casting["characters"]["sadie"] def test_promotion_sets_original_status(self, sample_image, tmp_project): data = promote_to_canonical( sample_image, asset_type="characters", entity_id="sadie", project_dir=tmp_project, ) assert data["status"] == "canonical" assert "promoted_to" in data def test_promotion_invalid_type_raises(self, sample_image, tmp_project): with pytest.raises(ValueError, match="Unknown asset_type"): promote_to_canonical( sample_image, asset_type="widgets", entity_id="foo", project_dir=tmp_project, ) def test_promotion_singular_type_works(self, sample_image, tmp_project): """Test that 'character' (singular) is accepted as well as 'characters'.""" promote_to_canonical( sample_image, asset_type="character", entity_id="sadie", project_dir=tmp_project, ) hero = ( tmp_project / "assets" / "char" / "sadie" / "base" / "sadie_identity.jpeg" ) assert hero.is_file() # ── Archive / Restore Tests ─────────────────────────────────── class TestArchive: def test_archive_moves_file(self, sample_image, tmp_project): original = Path(str(sample_image)) # copy path before move dest = archive_with_sidecar(sample_image, tmp_project) assert dest.is_file() assert not original.is_file() assert "_archive" in str(dest) def test_archive_moves_sidecar(self, sample_image, tmp_project): write_sidecar(sample_image, {"notes": "test archive"}) dest = archive_with_sidecar(sample_image, tmp_project) sc = _sidecar_path(dest) assert sc.is_file() data = json.loads(sc.read_text()) assert data["status"] == "archived" assert data["notes"] == "test archive" def test_archive_sets_archived_from(self, sample_image, tmp_project): dest = archive_with_sidecar(sample_image, tmp_project) sc_data = read_sidecar(dest) assert sc_data is not None assert "archived_from" in sc_data class TestRestore: def test_restore_from_archive(self, sample_image, tmp_project): # Archive first dest = archive_with_sidecar(sample_image, tmp_project) # Then restore restored = restore_from_archive(dest, tmp_project) assert restored.is_file() assert "_archive" not in str(restored.relative_to(tmp_project)) def test_restore_sets_candidate_status(self, sample_image, tmp_project): dest = archive_with_sidecar(sample_image, tmp_project) restored = restore_from_archive(dest, tmp_project) sc_data = read_sidecar(restored) assert sc_data is not None assert sc_data["status"] == "candidate" assert "archived_from" not in sc_data def test_restore_moves_sidecar(self, sample_image, tmp_project): write_sidecar(sample_image, {"notes": "round trip"}) dest = archive_with_sidecar(sample_image, tmp_project) restored = restore_from_archive(dest, tmp_project) sc_data = read_sidecar(restored) assert sc_data is not None assert sc_data["notes"] == "round trip" # ── Scanner Tests ───────────────────────────────────────────── class TestScanner: def test_finds_missing_sidecars(self, tmp_project): # Create some media files without sidecars output = tmp_project / "output" (output / "test1.jpg").write_bytes(b"\xff" * 10) (output / "test2.png").write_bytes(b"\x89PNG" + b"\x00" * 10) (output / "test3.mp4").write_bytes(b"\x00" * 10) missing = scan_for_missing_sidecars(output) assert len(missing) == 3 def test_ignores_files_with_sidecars(self, sample_image, tmp_project): write_sidecar(sample_image, {"status": "candidate"}) output = tmp_project / "output" missing = scan_for_missing_sidecars(output) paths = [str(p) for p in missing] assert str(sample_image) not in paths def test_ignores_meta_directories(self, tmp_project): output = tmp_project / "output" meta_img = ( output / "refs" / "_canonical" / "characters" / "sadie" / "_meta" / "hero.jpg" ) meta_img.parent.mkdir(parents=True, exist_ok=True) meta_img.write_bytes(b"\xff" * 10) missing = scan_for_missing_sidecars(output) paths = [str(p) for p in missing] assert str(meta_img) not in paths def test_auto_stub_creates_sidecars(self, tmp_project): output = tmp_project / "output" (output / "a.jpg").write_bytes(b"\xff" * 10) (output / "b.png").write_bytes(b"\x89" * 10) count = auto_stub_missing(output) assert count == 2 # Verify sidecars were created assert (output / "a.jpg.json").is_file() assert (output / "b.png.json").is_file() def test_auto_stub_idempotent(self, tmp_project): output = tmp_project / "output" (output / "a.jpg").write_bytes(b"\xff" * 10) count1 = auto_stub_missing(output) count2 = auto_stub_missing(output) assert count1 == 1 assert count2 == 0 # Already has sidecar # ── Pipeline Sidecar Tests ──────────────────────────────────── class TestPipelineSidecar: def test_write_pipeline_sidecar_RETIRED_replacement(self, sample_image): # R6 Phase 7 — migrated to populate_sidecar + write_sidecar_dict. sc_dict = populate_sidecar( receipt=None, payload={ "model": "seedream-v4.5", "modality": "image_t2i", "prompt": "Medium close-up of Sadie", "cost_usd": 0.039, }, gate_results={"black_frame": "pass"}, generation_params={"aspect_ratio": "9:16"}, shot_id="EP001_SH03", pipeline="image_t2i", ) sc_path = sample_image.with_suffix(sample_image.suffix + ".json") write_sidecar_dict(sc_path, sc_dict) assert sc_dict["source"] == "pipeline" assert sc_dict["status"] == "candidate" # Dual-emit: provenance.model is populated AND top-level model is populated. assert sc_dict["provenance"]["model"] == "seedream-v4.5" assert sc_dict["model"] == "seedream-v4.5" # Cost key-name split: provenance.cost vs top-level cost_usd. assert sc_dict["provenance"]["cost"] == 0.039 assert sc_dict["cost_usd"] == 0.039 assert sc_dict["provenance"]["shot_id"] == "EP001_SH03" def test_pipeline_sidecar_with_inputs_snapshot(self, sample_image): # R6 Phase 7 — inputs_snapshot_hash / location_id are now explicit kwargs. sc_dict = populate_sidecar( receipt=None, payload={ "model": "kling-v3", "modality": "video_i2v", "prompt": "Test prompt", }, inputs_snapshot_hash="abc123", location_id="int_apartment", shot_id="EP001_SH03", pipeline="video_i2v", refs_used=[{"role": "character", "id": "sadie", "display_name": "Sadie"}], ) assert sc_dict["provenance"]["inputs_snapshot_hash"] == "abc123" assert sc_dict["provenance"]["location_id"] == "int_apartment" assert len(sc_dict["provenance"]["refs_used"]) == 1 assert sc_dict["provenance"]["refs_used"][0]["id"] == "sadie" def test_pipeline_sidecar_persists(self, sample_image): # R6 Phase 7 — write through populate + write_sidecar_dict, read via read_sidecar. sc_dict = populate_sidecar( receipt=None, payload={"model": "test", "modality": "image_t2i", "prompt": "test"}, pipeline="image_t2i", ) sc_path = sample_image.with_suffix(sample_image.suffix + ".json") write_sidecar_dict(sc_path, sc_dict) result = read_sidecar(sample_image) assert result is not None assert result["source"] == "pipeline" # ── Edge Cases ──────────────────────────────────────────────── class TestEdgeCases: def test_nonexistent_directory_scanner(self): missing = scan_for_missing_sidecars(Path("/tmp/does_not_exist_abc123")) assert missing == [] def test_hidden_files_ignored(self, tmp_project): output = tmp_project / "output" (output / ".hidden.jpg").write_bytes(b"\xff" * 10) missing = scan_for_missing_sidecars(output) assert len(missing) == 0 def test_schema_version_always_set(self, sample_image): write_sidecar(sample_image, {"custom": "data"}) result = read_sidecar(sample_image) assert result["schema_version"] == SCHEMA_VERSION def test_updated_at_always_set(self, sample_image): write_sidecar(sample_image, {}) result = read_sidecar(sample_image) assert "updated_at" in result assert result["updated_at"].endswith("Z")