413 lines
15 KiB
Python
413 lines
15 KiB
Python
"""Rename plan building, validation, and export (US04-01).
|
|
|
|
Plans are built against a real temporary library so path validation (missing sources,
|
|
collisions, case-only, Unicode, root escapes, symlinks) is exercised against real
|
|
filesystem behaviour. Planning must never mutate anything: every test that builds a
|
|
plan also asserts the library is byte-for-byte unchanged.
|
|
"""
|
|
|
|
import json
|
|
import random
|
|
import unicodedata
|
|
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
|
|
from photo_pipeline.services.renames import (
|
|
BLOCKING_CODES,
|
|
RenameError,
|
|
RenameService,
|
|
plan_checksum,
|
|
)
|
|
|
|
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, names=("a.jpg", "b.jpg"), *, approved_name="2019 Rome"):
|
|
"""Create a real album folder with files, registered assets, and an approved
|
|
proposal naming it ``approved_name``."""
|
|
folder = lib / album
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
with sf() as session:
|
|
for name in names:
|
|
path = folder / name
|
|
path.write_bytes(b"x" * 32)
|
|
session.add(
|
|
Asset(
|
|
id=str(uuid.uuid4()),
|
|
original_path=str(path),
|
|
current_path=str(path),
|
|
discovered_at=NOW,
|
|
hash_version=1,
|
|
byte_size=32,
|
|
current_sha256="deadbeef",
|
|
)
|
|
)
|
|
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 _snapshot(lib):
|
|
"""Every path under the library plus its bytes — proof planning mutates nothing."""
|
|
return {
|
|
str(p.relative_to(lib)): (p.read_bytes() if p.is_file() else None)
|
|
for p in sorted(lib.rglob("*"))
|
|
}
|
|
|
|
|
|
def _service(sf, lib):
|
|
return RenameService(sf, library_roots=(lib,))
|
|
|
|
|
|
def _codes(plan):
|
|
return {issue["code"] for op in plan["operations"] for issue in op["issues"]}
|
|
|
|
|
|
# ── happy path ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_plan_lists_sources_destinations_assets_and_hashes(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
_album(sf, lib, "rome", approved_name="2019 Rome")
|
|
before = _snapshot(lib)
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
|
|
assert plan["state"] == "validated" and plan["applicable"] is True
|
|
assert plan["operation_count"] == 1 and plan["blockers"] == []
|
|
operation = plan["operations"][0]
|
|
assert operation["source_path"] == str(lib / "rome")
|
|
assert operation["destination_path"] == str(lib / "2019 Rome")
|
|
assert operation["asset_count"] == 2 and len(operation["asset_ids"]) == 2
|
|
assert set(operation["expected_sha256"].values()) == {"deadbeef"}
|
|
assert operation["journal_state"] == "planned" # journal untouched until apply
|
|
assert _snapshot(lib) == before, "planning must not touch the filesystem"
|
|
|
|
|
|
def test_plans_without_approved_proposals_are_refused(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
with pytest.raises(RenameError):
|
|
_service(sf, lib).build_plan()
|
|
|
|
|
|
# ── validation goldens ───────────────────────────────────────────────────────
|
|
|
|
|
|
def test_missing_source_folder_blocks_the_plan(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
folder = _album(sf, lib, "rome")
|
|
for child in folder.iterdir():
|
|
child.unlink()
|
|
folder.rmdir()
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
assert plan["state"] == "invalid" and plan["applicable"] is False
|
|
assert "source_missing" in _codes(plan)
|
|
|
|
|
|
def test_changed_source_bytes_block_the_plan(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
folder = _album(sf, lib, "rome")
|
|
(folder / "a.jpg").write_bytes(b"different length entirely")
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
assert "source_changed" in _codes(plan) and plan["state"] == "invalid"
|
|
|
|
|
|
def test_existing_destination_is_never_overwritten(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
_album(sf, lib, "rome", approved_name="2019 Rome")
|
|
(lib / "2019 Rome").mkdir() # something already lives there
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
assert "destination_exists" in _codes(plan) and plan["state"] == "invalid"
|
|
|
|
|
|
def test_two_albums_targeting_one_destination_collide(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
_album(sf, lib, "rome", approved_name="Same Name")
|
|
_album(sf, lib, "paris", names=("c.jpg",), approved_name="Same Name")
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
assert "duplicate_target" in _codes(plan) and plan["state"] == "invalid"
|
|
|
|
|
|
def test_destinations_differing_only_by_case_collide(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
_album(sf, lib, "rome", approved_name="Holiday")
|
|
_album(sf, lib, "paris", names=("c.jpg",), approved_name="holiday")
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
assert "unicode_collision" in _codes(plan) and plan["state"] == "invalid"
|
|
|
|
|
|
def test_destinations_differing_only_by_unicode_normalization_collide(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
composed = unicodedata.normalize("NFC", "Café")
|
|
decomposed = unicodedata.normalize("NFD", "Café")
|
|
assert composed != decomposed
|
|
_album(sf, lib, "rome", approved_name=composed)
|
|
_album(sf, lib, "paris", names=("c.jpg",), approved_name=decomposed)
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
assert "unicode_collision" in _codes(plan) and plan["state"] == "invalid"
|
|
|
|
|
|
def test_case_only_rename_is_flagged_but_not_blocking(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
_album(sf, lib, "rome", approved_name="Rome")
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
operation = plan["operations"][0]
|
|
assert operation["case_only"] is True
|
|
# A case-only rename needs the staged procedure, but it is not a blocker.
|
|
assert "destination_exists" not in _codes(plan)
|
|
assert plan["state"] == "validated"
|
|
|
|
|
|
def test_symlinked_source_is_refused(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
(outside / "a.jpg").write_bytes(b"x" * 32)
|
|
link = lib / "linked"
|
|
link.symlink_to(outside, target_is_directory=True)
|
|
|
|
with sf() as session:
|
|
session.add(
|
|
Asset(
|
|
id=str(uuid.uuid4()),
|
|
original_path=str(link / "a.jpg"),
|
|
current_path=str(link / "a.jpg"),
|
|
discovered_at=NOW,
|
|
hash_version=1,
|
|
byte_size=32,
|
|
)
|
|
)
|
|
session.add(
|
|
AlbumProposal(
|
|
id=str(uuid.uuid4()),
|
|
album="linked",
|
|
proposed_name="Linked",
|
|
final_name="Linked",
|
|
status="approved",
|
|
version=2,
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
assert "symlink" in _codes(plan) and plan["state"] == "invalid"
|
|
|
|
|
|
def test_paths_outside_every_library_root_are_refused(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
stray = tmp_path / "elsewhere" / "rome"
|
|
stray.mkdir(parents=True)
|
|
(stray / "a.jpg").write_bytes(b"x" * 32)
|
|
with sf() as session:
|
|
session.add(
|
|
Asset(
|
|
id=str(uuid.uuid4()),
|
|
original_path=str(stray / "a.jpg"),
|
|
current_path=str(stray / "a.jpg"),
|
|
discovered_at=NOW,
|
|
hash_version=1,
|
|
byte_size=32,
|
|
)
|
|
)
|
|
session.add(
|
|
AlbumProposal(
|
|
id=str(uuid.uuid4()),
|
|
album="rome",
|
|
proposed_name="Rome",
|
|
final_name="Rome",
|
|
status="approved",
|
|
version=2,
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
assert "root_escape" in _codes(plan) and plan["state"] == "invalid"
|
|
|
|
|
|
def test_excluded_source_is_refused(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
_album(sf, lib, "_IGNORE/private", approved_name="Private")
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
assert "excluded_path" in _codes(plan) and plan["state"] == "invalid"
|
|
|
|
|
|
# ── export ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_export_is_portable_versioned_and_credential_free(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
_album(sf, lib, "rome")
|
|
service = _service(sf, lib)
|
|
plan = service.build_plan()
|
|
|
|
exported = service.export(plan["id"])
|
|
assert exported["schema_version"] == 1
|
|
assert exported["checksum"] == plan["checksum"]
|
|
assert len(exported["operations"]) == 1
|
|
# Portable: round-trips through JSON unchanged.
|
|
assert json.loads(json.dumps(exported)) == exported
|
|
# No credentials or secrets anywhere in the document.
|
|
flat = json.dumps(exported).lower()
|
|
for secret in ("api_key", "apikey", "password", "token", "secret"):
|
|
assert secret not in flat
|
|
|
|
|
|
def test_export_of_an_unknown_plan_is_refused(tmp_path):
|
|
_, sf, lib = _factory(tmp_path)
|
|
with pytest.raises(RenameError):
|
|
_service(sf, lib).export("nope")
|
|
|
|
|
|
# ── properties ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_property_a_validated_plan_has_no_duplicate_destinations(tmp_path):
|
|
"""Across randomized name sets, any plan that validates is collision-free."""
|
|
rng = random.Random(4242)
|
|
names = ["Rome", "rome", "Café", unicodedata.normalize("NFD", "Café"), "Paris", "2019 Trip"]
|
|
for trial in range(25):
|
|
root = tmp_path / f"trial{trial}"
|
|
root.mkdir()
|
|
_, sf, lib = _factory(root)
|
|
chosen = [rng.choice(names) for _ in range(rng.randint(1, 3))]
|
|
for index, name in enumerate(chosen):
|
|
_album(sf, lib, f"album{index}", names=(f"{index}.jpg",), approved_name=name)
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
destinations = [op["destination_path"] for op in plan["operations"]]
|
|
if plan["state"] == "validated":
|
|
folded = [unicodedata.normalize("NFC", d).casefold() for d in destinations]
|
|
assert len(destinations) == len(set(destinations))
|
|
assert len(folded) == len(set(folded)), "validated plan has a normalized collision"
|
|
|
|
|
|
def test_property_a_plan_preserves_the_asset_set(tmp_path):
|
|
"""Planning never invents, drops, or duplicates an asset."""
|
|
rng = random.Random(99)
|
|
for trial in range(15):
|
|
root = tmp_path / f"trial{trial}"
|
|
root.mkdir()
|
|
_, sf, lib = _factory(root)
|
|
expected = set()
|
|
for index in range(rng.randint(1, 3)):
|
|
count = rng.randint(1, 4)
|
|
_album(
|
|
sf,
|
|
lib,
|
|
f"album{index}",
|
|
names=tuple(f"{n}.jpg" for n in range(count)),
|
|
approved_name=f"Album {index}",
|
|
)
|
|
with sf() as session:
|
|
expected = set(session.scalars(select(Asset.id)))
|
|
|
|
plan = _service(sf, lib).build_plan()
|
|
planned = [aid for op in plan["operations"] for aid in op["asset_ids"]]
|
|
assert sorted(planned) == sorted(set(planned)), "an asset appears twice in one plan"
|
|
assert set(planned) == expected
|
|
|
|
|
|
def test_checksum_is_stable_and_sensitive():
|
|
base = [{"sequence": 0, "source_path": "/a", "destination_path": "/b", "asset_ids": ["x", "y"]}]
|
|
same_order_swapped = [
|
|
{"sequence": 0, "source_path": "/a", "destination_path": "/b", "asset_ids": ["y", "x"]}
|
|
]
|
|
changed = [
|
|
{"sequence": 0, "source_path": "/a", "destination_path": "/c", "asset_ids": ["x", "y"]}
|
|
]
|
|
assert plan_checksum(base) == plan_checksum(same_order_swapped)
|
|
assert plan_checksum(base) != plan_checksum(changed)
|
|
|
|
|
|
def test_blocking_codes_are_the_documented_set():
|
|
# Guards against a new issue code silently becoming non-blocking.
|
|
assert "cross_filesystem" not in BLOCKING_CODES
|
|
assert "case_only" not in BLOCKING_CODES
|
|
assert {"source_missing", "destination_exists", "root_escape", "symlink"} <= BLOCKING_CODES
|
|
|
|
|
|
# ── API ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_api_build_get_list_and_export(tmp_path):
|
|
config, sf, lib = _factory(tmp_path)
|
|
_album(sf, lib, "rome")
|
|
before = _snapshot(lib)
|
|
|
|
with TestClient(create_app(config)) as client:
|
|
built = client.post("/api/v1/rename-plans").json()
|
|
assert built["state"] == "validated"
|
|
plan_id = built["id"]
|
|
|
|
fetched = client.get(f"/api/v1/rename-plans/{plan_id}").json()
|
|
assert fetched["checksum"] == built["checksum"]
|
|
|
|
listed = client.get("/api/v1/rename-plans").json()
|
|
assert listed["total"] == 1 and listed["items"][0]["id"] == plan_id
|
|
|
|
exported = client.get(f"/api/v1/rename-plans/{plan_id}/export").json()
|
|
assert exported["plan_id"] == plan_id and exported["schema_version"] == 1
|
|
|
|
assert client.get("/api/v1/rename-plans/nope").status_code == 404
|
|
|
|
assert _snapshot(lib) == before, "the plan API must not touch the filesystem"
|
|
|
|
|
|
def test_api_refuses_to_plan_without_approved_proposals(tmp_path):
|
|
config, sf, lib = _factory(tmp_path)
|
|
with TestClient(create_app(config)) as client:
|
|
response = client.post("/api/v1/rename-plans")
|
|
assert response.status_code == 422
|
|
assert response.json()["error"]["code"] == "nothing_to_plan"
|
|
|
|
|
|
def test_plan_survives_restart(tmp_path):
|
|
config, sf, lib = _factory(tmp_path)
|
|
_album(sf, lib, "rome")
|
|
with TestClient(create_app(config)) as client:
|
|
plan_id = client.post("/api/v1/rename-plans").json()["id"]
|
|
|
|
with TestClient(create_app(config)) as client:
|
|
after = client.get(f"/api/v1/rename-plans/{plan_id}").json()
|
|
assert after["state"] == "validated" and after["operation_count"] == 1
|