"""Reconciliation: stable identity across move/copy/replace/remove, with idempotent rescans and durable state after a restart. Fixtures live here rather than in a sibling conftest.py: the characterization suite imports its own ``conftest`` by bare name, and a second sibling conftest would shadow it under pytest's prepend import mode. """ import shutil import numpy as np import pytest from PIL import Image from sqlalchemy import select from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations from photo_pipeline.models import Asset, AssetPath from photo_pipeline.services.inventory import InventoryService, Occurrence @pytest.fixture def db_url(tmp_path): url = f"sqlite:///{tmp_path / 'inv.db'}" run_migrations(url) return url @pytest.fixture def make_factory(): engines = [] def _make(url): engine = create_db_engine(url) engines.append(engine) return create_session_factory(engine) yield _make for engine in engines: engine.dispose() @pytest.fixture def inventory(db_url, make_factory): return InventoryService(make_factory(db_url)) @pytest.fixture def make_image(): def _make(path, seed=1, size=(64, 48)): path.parent.mkdir(parents=True, exist_ok=True) rng = np.random.default_rng(seed) arr = rng.integers(0, 256, (size[1], size[0], 3), dtype=np.uint8) Image.fromarray(arr).save(path, quality=95) return path return _make @pytest.fixture def assets_by_id(): def _snapshot(factory): with factory() as session: return {a.id: a for a in session.execute(select(Asset)).scalars()} return _snapshot def test_initial_scan_registers_new_assets(tmp_path, inventory, make_image): lib = tmp_path / "lib" make_image(lib / "a.jpg", seed=1) make_image(lib / "album" / "b.jpg", seed=2) result = inventory.scan(lib) assert result.counts.get(Occurrence.NEW.value) == 2 assert set(result.occurrences.values()) == {Occurrence.NEW.value} assert len(result.asset_ids) == 2 def test_rescan_unchanged_is_idempotent( tmp_path, inventory, db_url, make_factory, make_image, assets_by_id ): lib = tmp_path / "lib" make_image(lib / "a.jpg", seed=1) make_image(lib / "b.jpg", seed=2) first = inventory.scan(lib) before = assets_by_id(make_factory(db_url)) second = inventory.scan(lib) after = assets_by_id(make_factory(db_url)) assert set(second.occurrences.values()) == {Occurrence.UNCHANGED.value} assert set(before) == set(after) # same asset ids, none created assert first.asset_ids == second.asset_ids def test_move_preserves_identity(tmp_path, inventory, db_url, make_factory, make_image): lib = tmp_path / "lib" src = make_image(lib / "a.jpg", seed=1) original_id = next(iter(inventory.scan(lib).asset_ids.values())) dst = lib / "album" / "moved.jpg" dst.parent.mkdir(parents=True, exist_ok=True) shutil.move(str(src), str(dst)) result = inventory.scan(lib) assert result.occurrences[str(dst)] == Occurrence.MOVED.value assert result.asset_ids[str(dst)] == original_id # id retained across the move # Path history: old occurrence closed, new occurrence open. with make_factory(db_url)() as session: paths = session.query(AssetPath).filter_by(asset_id=original_id).all() by_path = {p.path: p for p in paths} assert by_path[str(src)].valid_until is not None assert by_path[str(dst)].valid_until is None def test_copy_creates_distinct_identity(tmp_path, inventory, make_image): lib = tmp_path / "lib" src = make_image(lib / "a.jpg", seed=1) first = inventory.scan(lib) original_id = first.asset_ids[str(src)] copy = lib / "a_copy.jpg" shutil.copy2(str(src), str(copy)) result = inventory.scan(lib) assert result.occurrences[str(copy)] == Occurrence.COPIED.value assert result.asset_ids[str(copy)] != original_id assert result.asset_ids[str(src)] == original_id def test_replace_versus_metadata_change(tmp_path, inventory, make_image): lib = tmp_path / "lib" same = make_image(lib / "same.jpg", seed=1) other = make_image(lib / "other.jpg", seed=2) ids = inventory.scan(lib).asset_ids # Metadata-style edit: append trailing bytes (pixels unchanged). with open(same, "ab") as handle: handle.write(b"\xff\xfe trailing") # Content replacement: overwrite with different pixels. make_image(other, seed=99) result = inventory.scan(lib) assert result.occurrences[str(same)] == Occurrence.METADATA_CHANGED.value assert result.occurrences[str(other)] == Occurrence.REPLACED.value # Identity survives both; hashes refreshed. assert result.asset_ids[str(same)] == ids[str(same)] assert result.asset_ids[str(other)] == ids[str(other)] def test_missing_file_is_flagged_not_deleted( tmp_path, inventory, db_url, make_factory, make_image, assets_by_id ): lib = tmp_path / "lib" a = make_image(lib / "a.jpg", seed=1) make_image(lib / "b.jpg", seed=2) ids = inventory.scan(lib).asset_ids missing_id = ids[str(a)] a.unlink() result = inventory.scan(lib) assert result.occurrences[str(a)] == Occurrence.MISSING.value assets = assets_by_id(make_factory(db_url)) assert missing_id in assets # not pruned assert assets[missing_id].missing_at is not None assert assets[missing_id].availability_state == "active" def test_reappearing_file_clears_missing( tmp_path, inventory, db_url, make_factory, make_image, assets_by_id ): lib = tmp_path / "lib" a = make_image(lib / "a.jpg", seed=1) ids = inventory.scan(lib).asset_ids asset_id = ids[str(a)] backup = tmp_path / "backup.jpg" shutil.move(str(a), str(backup)) inventory.scan(lib) # marks missing shutil.move(str(backup), str(a)) # reappears at the same path inventory.scan(lib) assets = assets_by_id(make_factory(db_url)) assert assets[asset_id].missing_at is None def test_identity_and_state_durable_across_restart( tmp_path, db_url, make_factory, make_image, assets_by_id ): from photo_pipeline.services.inventory import InventoryService lib = tmp_path / "lib" src = make_image(lib / "a.jpg", seed=1) service_a = InventoryService(make_factory(db_url)) original_id = service_a.scan(lib).asset_ids[str(src)] # "Restart": a brand-new service/session factory against the same database. dst = lib / "renamed.jpg" shutil.move(str(src), str(dst)) service_b = InventoryService(make_factory(db_url)) result = service_b.scan(lib) assert result.asset_ids[str(dst)] == original_id assert result.occurrences[str(dst)] == Occurrence.MOVED.value assets = assets_by_id(make_factory(db_url)) assert assets[original_id].current_path == str(dst)