"""Applying and verifying guarded renames (US04-03). Everything here runs against a real temporary library: files really move, and the tests assert the bytes survived, the stable asset IDs kept their identity, the path history was recorded, and nothing unexpected was ever overwritten. """ import uuid from datetime import datetime, timezone import pytest from fastapi.testclient import TestClient from sqlalchemy import select from photo_pipeline.api.app import create_app from photo_pipeline.config import Config from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations from photo_pipeline.models import AlbumProposal, Asset, AssetPath from photo_pipeline.services import hashing from photo_pipeline.services.rename_apply import ( ApplyConflict, ApplyError, RenameApplyService, ) from photo_pipeline.services.rename_journal import JournalState, RenameJournal from photo_pipeline.services.renames import RenameService NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) def _factory(tmp_path): (tmp_path / "data").mkdir() lib = tmp_path / "lib" lib.mkdir() config = Config.from_env( { "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib), } ) run_migrations(config.database_url) return config, create_session_factory(create_db_engine(config.database_url)), lib def _album(sf, lib, album, *, approved_name, names=("a.jpg", "b.jpg")): folder = lib / album folder.mkdir(parents=True, exist_ok=True) with sf() as session: for index, name in enumerate(names): path = folder / name path.write_bytes(f"content-{album}-{index}".encode()) session.add( Asset( id=str(uuid.uuid4()), original_path=str(path), current_path=str(path), discovered_at=NOW, hash_version=1, byte_size=path.stat().st_size, current_sha256=hashing.sha256_file(path), ) ) session.add( AlbumProposal( id=str(uuid.uuid4()), album=album, proposed_name=approved_name, final_name=approved_name, status="approved", version=2, ) ) session.commit() return folder def _plan(sf, lib): return RenameService(sf, library_roots=(lib,)).build_plan() def _apply(sf, lib, plan, **kwargs): service = RenameApplyService(sf, library_roots=(lib,)) return service.apply(plan["id"], expected_version=plan["version"], **kwargs) def _paths(sf): with sf() as session: return {a.id: a.current_path for a in session.scalars(select(Asset))} def _bytes_under(root): return { str(p.relative_to(root)): p.read_bytes() for p in sorted(root.rglob("*")) if p.is_file() } # ── normal apply ───────────────────────────────────────────────────────────── def test_apply_moves_the_folder_and_preserves_bytes_and_identity(tmp_path): _, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="2019 Rome") before_ids = set(_paths(sf)) before_bytes = _bytes_under(lib) plan = _plan(sf, lib) result = _apply(sf, lib, plan) assert result["applied"] == 1 and result["failed"] == 0 assert result["state"] == "applied" assert (lib / "2019 Rome").is_dir() and not (lib / "rome").exists() # Same asset IDs, new paths, identical bytes. after = _paths(sf) assert set(after) == before_ids, "renaming must not change asset identity" assert all("2019 Rome" in path for path in after.values()) assert _bytes_under(lib) == { key.replace("rome/", "2019 Rome/"): value for key, value in before_bytes.items() } def test_apply_records_the_path_history(tmp_path): _, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",)) plan = _plan(sf, lib) _apply(sf, lib, plan) with sf() as session: rows = list(session.scalars(select(AssetPath).order_by(AssetPath.valid_from))) open_rows = [r for r in rows if r.valid_until is None] assert len(open_rows) == 1 and "2019 Rome" in open_rows[0].path assert open_rows[0].reason == "rename" def test_every_operation_reaches_complete_in_the_journal(tmp_path): _, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",)) plan = _plan(sf, lib) _apply(sf, lib, plan) states = {op["journal_state"] for op in RenameJournal(sf).operations(plan["id"])} assert states == {JournalState.COMPLETE} # ── confirmation and leases ────────────────────────────────────────────────── def test_apply_rejects_a_stale_version_without_touching_disk(tmp_path): _, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="2019 Rome") plan = _plan(sf, lib) before = _bytes_under(lib) service = RenameApplyService(sf, library_roots=(lib,)) with pytest.raises(ApplyConflict): service.apply(plan["id"], expected_version=plan["version"] + 5) assert _bytes_under(lib) == before and (lib / "rome").exists() def test_apply_rejects_a_changed_checksum(tmp_path): _, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="2019 Rome") plan = _plan(sf, lib) service = RenameApplyService(sf, library_roots=(lib,)) with pytest.raises(ApplyConflict): service.apply( plan["id"], expected_version=plan["version"], expected_checksum="not-the-checksum" ) def test_an_invalid_plan_can_never_be_applied(tmp_path): _, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="2019 Rome") (lib / "2019 Rome").mkdir() # destination occupied → plan is invalid plan = _plan(sf, lib) assert plan["state"] == "invalid" service = RenameApplyService(sf, library_roots=(lib,)) with pytest.raises(ApplyError): service.apply(plan["id"], expected_version=plan["version"]) def test_repeated_apply_is_a_no_op(tmp_path): _, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="2019 Rome") plan = _plan(sf, lib) _apply(sf, lib, plan) after_first = _bytes_under(lib) service = RenameApplyService(sf, library_roots=(lib,)) fresh_version = RenameService(sf, library_roots=(lib,)).get(plan["id"])["version"] second = service.apply(plan["id"], expected_version=fresh_version) assert second["applied"] == 0 and second["skipped"] == 1 assert _bytes_under(lib) == after_first # ── preconditions rechecked at mutation time ───────────────────────────────── def test_a_source_changed_after_planning_is_refused(tmp_path): _, sf, lib = _factory(tmp_path) folder = _album(sf, lib, "rome", approved_name="2019 Rome") plan = _plan(sf, lib) # The file changes between preview and confirmation. (folder / "a.jpg").write_bytes(b"tampered") result = _apply(sf, lib, plan) assert result["failed"] == 1 and result["applied"] == 0 assert (lib / "rome").exists(), "a failed precondition must leave the source alone" operation = RenameJournal(sf).operations(plan["id"])[0] assert operation["journal_state"] == JournalState.FAILED assert operation["error_code"] == "source_changed" def test_an_occupant_appearing_after_planning_is_never_overwritten(tmp_path): _, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="2019 Rome") plan = _plan(sf, lib) # Someone creates the destination between validation and apply. (lib / "2019 Rome").mkdir() (lib / "2019 Rome" / "precious.jpg").write_bytes(b"do not lose me") result = _apply(sf, lib, plan) assert result["failed"] == 1 assert (lib / "2019 Rome" / "precious.jpg").read_bytes() == b"do not lose me" assert (lib / "rome").exists() # ── staged procedures ──────────────────────────────────────────────────────── def test_case_only_rename_uses_the_staged_procedure(tmp_path): _, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="Rome", names=("a.jpg",)) plan = _plan(sf, lib) assert plan["operations"][0]["case_only"] is True result = _apply(sf, lib, plan) assert result["applied"] == 1 # The directory entry now carries the new casing, and no staging dir survives. entries = {p.name for p in lib.iterdir()} assert "Rome" in entries assert not any(name.startswith(".rename-") for name in entries) assert list(_paths(sf).values())[0].endswith("Rome/a.jpg") def test_cross_filesystem_move_copies_verifies_then_deletes(tmp_path): """The cross-filesystem branch is driven directly: a real second filesystem is not portable in CI, but the copy-verify-delete procedure is what matters.""" _, sf, lib = _factory(tmp_path) source = lib / "rome" source.mkdir() (source / "a.jpg").write_bytes(b"payload") destination = lib / "2019 Rome" RenameApplyService(sf, library_roots=(lib,))._copy_verify_delete(source, destination) assert not source.exists() assert (destination / "a.jpg").read_bytes() == b"payload" assert not any(p.name.startswith(".rename-") for p in lib.iterdir()) # ── restart durability ─────────────────────────────────────────────────────── def test_applied_state_survives_a_restart(tmp_path): config, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",)) plan = _plan(sf, lib) _apply(sf, lib, plan) expected = _paths(sf) reopened = create_session_factory(create_db_engine(config.database_url)) assert _paths(reopened) == expected states = {op["journal_state"] for op in RenameJournal(reopened).operations(plan["id"])} assert states == {JournalState.COMPLETE} # ── API ────────────────────────────────────────────────────────────────────── def test_api_apply_requires_the_current_version(tmp_path): config, sf, lib = _factory(tmp_path) _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",)) with TestClient(create_app(config)) as client: plan = client.post("/api/v1/rename-plans").json() stale = client.post( f"/api/v1/rename-plans/{plan['id']}/apply", json={"expected_version": plan["version"] + 9}, ) assert stale.status_code == 409 assert stale.json()["error"]["code"] == "version_conflict" assert (lib / "rome").exists() applied = client.post( f"/api/v1/rename-plans/{plan['id']}/apply", json={"expected_version": plan["version"], "expected_checksum": plan["checksum"]}, ).json() assert applied["applied"] == 1 and applied["state"] == "applied" assert (lib / "2019 Rome").is_dir() recovery = client.get("/api/v1/rename-recovery").json() assert recovery["blocks_mutation"] is False and recovery["items"] == []