From c7a616f4b55fa2b3ba7cf67716cd719e200e5085 Mon Sep 17 00:00:00 2001 From: domverse Date: Sat, 15 Aug 2026 14:51:47 +0200 Subject: [PATCH] US04-01: Build and Export Rename Plans (#66) --- migrations/versions/0007_rename_plans.py | 100 ++++++ photo_pipeline/api/app.py | 2 + photo_pipeline/api/routes/renames.py | 54 +++ photo_pipeline/models/__init__.py | 3 + photo_pipeline/models/renames.py | 84 +++++ photo_pipeline/services/renames.py | 426 +++++++++++++++++++++++ tests/integration/test_rename_plans.py | 412 ++++++++++++++++++++++ tests/story_traceability.json | 3 + 8 files changed, 1084 insertions(+) create mode 100644 migrations/versions/0007_rename_plans.py create mode 100644 photo_pipeline/api/routes/renames.py create mode 100644 photo_pipeline/models/renames.py create mode 100644 photo_pipeline/services/renames.py create mode 100644 tests/integration/test_rename_plans.py diff --git a/migrations/versions/0007_rename_plans.py b/migrations/versions/0007_rename_plans.py new file mode 100644 index 0000000..395564d --- /dev/null +++ b/migrations/versions/0007_rename_plans.py @@ -0,0 +1,100 @@ +"""Rename plans and their per-operation journal (US04-01, US04-02). + +Revision ID: 0007_rename_plans +Revises: 0006_album_proposals +Create Date: 2026-08-15 + +The journal columns (state machine, attempts, fencing token, verification evidence) +are created here alongside the plan itself so the crash-safe journal of US04-02 does +not have to re-migrate the very rows US04-01 creates. US04-01 fills the planning and +validation columns; the journal columns stay at their initial values until apply. +""" + +import sqlalchemy as sa +from alembic import op + +revision = "0007_rename_plans" +down_revision = "0006_album_proposals" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "rename_plans", + sa.Column("id", sa.String(), primary_key=True), + # planned | validated | invalid | applying | applied | failed | rolled_back + sa.Column("state", sa.String(), nullable=False, server_default="planned"), + sa.Column("schema_version", sa.Integer(), nullable=False, server_default="1"), + # Content checksum over the plan's operations; a portable export carries it + # so a stale confirmation can be detected without trusting the browser. + sa.Column("checksum", sa.String(), nullable=True), + sa.Column("blockers", sa.String(), nullable=True), # JSON array of issues + sa.Column("operation_count", sa.Integer(), nullable=False, server_default="0"), + # Optimistic concurrency for confirmation (apply must present this). + sa.Column("version", sa.Integer(), nullable=False, server_default="1"), + sa.Column("validated_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("applied_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + ) + op.create_index("ix_rename_plans_state", "rename_plans", ["state"]) + + op.create_table( + "rename_operations", + sa.Column("id", sa.String(), primary_key=True), + sa.Column( + "plan_id", + sa.String(), + sa.ForeignKey("rename_plans.id", ondelete="CASCADE"), + nullable=False, + ), + # Deterministic order within the plan (parents before children). + sa.Column("sequence", sa.Integer(), nullable=False), + # move_folder | move_file — folder moves are the default (album renames). + sa.Column("operation", sa.String(), nullable=False), + sa.Column("album", sa.String(), nullable=True), + sa.Column("source_path", sa.String(), nullable=False), + sa.Column("destination_path", sa.String(), nullable=False), + # Preconditions captured at planning time and rechecked before mutation. + sa.Column("asset_ids", sa.String(), nullable=True), # JSON array + sa.Column("expected_sha256", sa.String(), nullable=True), # JSON {asset_id: sha} + sa.Column("asset_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("same_filesystem", sa.Boolean(), nullable=True), + sa.Column("case_only", sa.Boolean(), nullable=False, server_default="0"), + sa.Column("issues", sa.String(), nullable=True), # JSON array of issues + # ── journal (US04-02): untouched until apply ───────────────────────── + # planned → validated → moving → moved → database_updated → verified + # → complete, or failed / rollback_required + sa.Column("journal_state", sa.String(), nullable=False, server_default="planned"), + sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("fencing_token", sa.Integer(), nullable=True), + sa.Column("worker_id", sa.String(), nullable=True), + sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("error_code", sa.String(), nullable=True), + sa.Column("error_message", sa.String(), nullable=True), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + sa.UniqueConstraint("plan_id", "sequence", name="uq_rename_operations_plan_sequence"), + ) + op.create_index("ix_rename_operations_plan_id", "rename_operations", ["plan_id"]) + op.create_index("ix_rename_operations_journal_state", "rename_operations", ["journal_state"]) + + +def downgrade() -> None: + op.drop_table("rename_operations") + op.drop_table("rename_plans") diff --git a/photo_pipeline/api/app.py b/photo_pipeline/api/app.py index 2462d37..742df90 100644 --- a/photo_pipeline/api/app.py +++ b/photo_pipeline/api/app.py @@ -22,6 +22,7 @@ from photo_pipeline.api.routes import ( inventory, jobs, library, + renames, safety, thumbnails, workflow, @@ -65,6 +66,7 @@ def create_app(config: Config | None = None) -> FastAPI: app.include_router(analysis.router, prefix="/api/v1") app.include_router(library.router, prefix="/api/v1") app.include_router(albums.router, prefix="/api/v1") + app.include_router(renames.router, prefix="/api/v1") # Static single-page app (hash-routed). Mounted last so /api/v1 wins. if FRONTEND_DIR.is_dir(): app.mount("/app", StaticFiles(directory=FRONTEND_DIR, html=True), name="app") diff --git a/photo_pipeline/api/routes/renames.py b/photo_pipeline/api/routes/renames.py new file mode 100644 index 0000000..5b7f74a --- /dev/null +++ b/photo_pipeline/api/routes/renames.py @@ -0,0 +1,54 @@ +"""Rename plan API (US04-01): build, inspect, and export plans. + +Planning is read-only — it previews and validates but never touches the filesystem. +Applying a plan is a later story, so there is deliberately no apply endpoint here. +""" + +from __future__ import annotations + +from fastapi import APIRouter, Request +from fastapi.responses import JSONResponse + +from photo_pipeline.services.renames import RenameError, RenameService + +router = APIRouter(tags=["renames"]) + + +def _service(request: Request) -> RenameService: + return RenameService( + request.app.state.session_factory, + library_roots=tuple(request.app.state.config.library_roots), + ) + + +def _error(status: int, code: str, message: str) -> JSONResponse: + return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}}) + + +@router.post("/rename-plans") +def build_plan(request: Request): + try: + return _service(request).build_plan() + except RenameError as error: + return _error(422, "nothing_to_plan", str(error)) + + +@router.get("/rename-plans") +def list_plans(request: Request) -> dict: + return _service(request).list_plans() + + +@router.get("/rename-plans/{plan_id}") +def get_plan(plan_id: str, request: Request): + plan = _service(request).get(plan_id) + if plan is None: + return _error(404, "not_found", f"unknown rename plan {plan_id}") + return plan + + +@router.get("/rename-plans/{plan_id}/export") +def export_plan(plan_id: str, request: Request): + try: + return _service(request).export(plan_id) + except RenameError as error: + return _error(404, "not_found", str(error)) diff --git a/photo_pipeline/models/__init__.py b/photo_pipeline/models/__init__.py index 17ca18a..983b437 100644 --- a/photo_pipeline/models/__init__.py +++ b/photo_pipeline/models/__init__.py @@ -12,6 +12,7 @@ from photo_pipeline.models.duplicates import ( DuplicateNegativeLink, ) from photo_pipeline.models.jobs import Job, JobEvent, JobItem +from photo_pipeline.models.renames import RenameOperation, RenamePlan from photo_pipeline.models.thumbnails import Thumbnail from photo_pipeline.models.workflow import AnalysisResult, SafetyReview @@ -25,6 +26,8 @@ __all__ = [ "Job", "JobItem", "JobEvent", + "RenamePlan", + "RenameOperation", "Thumbnail", "SafetyReview", "AnalysisResult", diff --git a/photo_pipeline/models/renames.py b/photo_pipeline/models/renames.py new file mode 100644 index 0000000..563ab66 --- /dev/null +++ b/photo_pipeline/models/renames.py @@ -0,0 +1,84 @@ +"""Rename plan and per-operation journal persistence (US04-01, US04-02). + +A plan is the complete, reviewable preview of a set of folder renames: every source, +destination, affected asset, expected hash, and validation issue. Nothing here +touches the filesystem — applying a plan is US04-03. + +The journal columns on ``RenameOperation`` belong to US04-02's crash-safe state +machine and are created with the table so the journal never has to re-migrate the +rows the planner writes. They stay at their defaults until apply runs. +""" + +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import ( + Boolean, + DateTime, + ForeignKey, + Integer, + String, + UniqueConstraint, + func, +) +from sqlalchemy.orm import Mapped, mapped_column + +from photo_pipeline.db import Base + + +class RenamePlan(Base): + __tablename__ = "rename_plans" + + id: Mapped[str] = mapped_column(String, primary_key=True) + # planned | validated | invalid | applying | applied | failed | rolled_back + state: Mapped[str] = mapped_column(String, nullable=False, default="planned") + schema_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + checksum: Mapped[str | None] = mapped_column(String) + blockers: Mapped[str | None] = mapped_column(String) # JSON array + operation_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + version: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + validated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + applied_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() + ) + + +class RenameOperation(Base): + __tablename__ = "rename_operations" + __table_args__ = ( + UniqueConstraint("plan_id", "sequence", name="uq_rename_operations_plan_sequence"), + ) + + id: Mapped[str] = mapped_column(String, primary_key=True) + plan_id: Mapped[str] = mapped_column( + ForeignKey("rename_plans.id", ondelete="CASCADE"), nullable=False, index=True + ) + sequence: Mapped[int] = mapped_column(Integer, nullable=False) + operation: Mapped[str] = mapped_column(String, nullable=False) + album: Mapped[str | None] = mapped_column(String) + source_path: Mapped[str] = mapped_column(String, nullable=False) + destination_path: Mapped[str] = mapped_column(String, nullable=False) + + asset_ids: Mapped[str | None] = mapped_column(String) # JSON array + expected_sha256: Mapped[str | None] = mapped_column(String) # JSON {asset_id: sha} + asset_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + same_filesystem: Mapped[bool | None] = mapped_column(Boolean) + case_only: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + issues: Mapped[str | None] = mapped_column(String) # JSON array + + # ── journal (US04-02) ──────────────────────────────────────────────────── + journal_state: Mapped[str] = mapped_column(String, nullable=False, default="planned") + attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + fencing_token: Mapped[int | None] = mapped_column(Integer) + worker_id: Mapped[str | None] = mapped_column(String) + verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + error_code: Mapped[str | None] = mapped_column(String) + error_message: Mapped[str | None] = mapped_column(String) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() + ) diff --git a/photo_pipeline/services/renames.py b/photo_pipeline/services/renames.py new file mode 100644 index 0000000..53d08b0 --- /dev/null +++ b/photo_pipeline/services/renames.py @@ -0,0 +1,426 @@ +"""RenameService — build, validate, and export guarded rename plans (US04-01). + +A plan turns approved album proposals into an itemized, reviewable preview: for every +album, the source folder, the destination folder, the assets that move with it, their +expected hashes, and every validation issue found. **Nothing here touches the +filesystem** — it only reads to validate. Applying a plan is US04-03. + +Validation is the point of the story, so every refusal is a structured issue rather +than an exception (concept §7 "guarded rename plan"): + +- ``source_missing`` — the registered folder is gone +- ``source_changed`` — an asset's bytes changed since inventory +- ``duplicate_target`` — two operations claim the same destination +- ``destination_exists`` — an unexpected path already occupies the destination +- ``case_only`` — informational: needs the staged procedure on + case-insensitive filesystems (not a blocker) +- ``unicode_collision`` — two destinations differ only by Unicode normalization +- ``root_escape`` — a path resolves outside the configured library +- ``excluded_path`` — a path lies under ``_IGNORE/`` +- ``symlink`` — source or destination is a symlink +- ``cross_filesystem`` — destination is on another device (copy-verify-delete) + +A plan with any blocking issue is ``invalid`` and can never be applied. Issues that +are purely informational (``case_only``, ``cross_filesystem``) are recorded on the +operation so the apply stage picks the right procedure. + +The JSON export is portable evidence: schema version, checksum, and the full operation +list, with no credentials and no absolute host paths beyond the library itself. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import unicodedata +import uuid +from datetime import datetime, timezone +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.orm import sessionmaker + +from photo_pipeline import path_policy +from photo_pipeline.models import AlbumProposal, Asset, RenameOperation, RenamePlan +from photo_pipeline.services.albums import album_label + +EXPORT_SCHEMA_VERSION = 1 +MOVE_FOLDER = "move_folder" + +# Issues that make a plan unapplyable. Everything else is advisory. +BLOCKING_CODES = frozenset( + { + "source_missing", + "source_changed", + "duplicate_target", + "destination_exists", + "unicode_collision", + "root_escape", + "excluded_path", + "symlink", + "no_library_root", + } +) + + +class RenameError(RuntimeError): + """Invalid request against a plan (unknown id, nothing to plan).""" + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _issue(code: str, message: str) -> dict: + return {"code": code, "message": message} + + +def _nfc(value: str) -> str: + return unicodedata.normalize("NFC", value) + + +def plan_checksum(operations: list[dict]) -> str: + """Content hash over the operations that define a plan. Two plans with the same + moves have the same checksum, so a stale confirmation is detectable.""" + payload = json.dumps( + [ + { + "sequence": op["sequence"], + "source_path": op["source_path"], + "destination_path": op["destination_path"], + "asset_ids": sorted(op["asset_ids"]), + } + for op in sorted(operations, key=lambda op: op["sequence"]) + ], + ensure_ascii=False, + sort_keys=True, + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +class RenameService: + def __init__(self, session_factory: sessionmaker, *, library_roots: tuple = ()) -> None: + self._session_factory = session_factory + self._roots = tuple(Path(root) for root in library_roots) + + # ── planning ────────────────────────────────────────────────────────────── + + def build_plan(self) -> dict: + """Build a plan from every approved album proposal. Read-only.""" + with self._session_factory() as session: + approved = list( + session.scalars( + select(AlbumProposal) + .where(AlbumProposal.status == "approved") + .order_by(AlbumProposal.album) + ) + ) + if not approved: + raise RenameError("no approved album proposals to plan") + assets = list( + session.scalars( + select(Asset).where( + Asset.canonical_asset_id.is_(None), + Asset.availability_state == "active", + Asset.current_path.is_not(None), + ) + ) + ) + + by_album: dict[str, list[Asset]] = {} + for asset in assets: + by_album.setdefault(album_label(asset.current_path, self._roots), []).append(asset) + + operations: list[dict] = [] + for sequence, proposal in enumerate(approved): + members = sorted(by_album.get(proposal.album, []), key=lambda a: a.current_path) + operations.append(self._operation(sequence, proposal, members)) + + self._detect_cross_operation_issues(operations) + return self._persist(operations) + + def _operation(self, sequence: int, proposal: AlbumProposal, members: list[Asset]) -> dict: + final_name = proposal.final_name or proposal.proposed_name or "" + issues: list[dict] = [] + + source = self._source_folder(members) + destination = source.parent / final_name if source else None + + operation = { + "sequence": sequence, + "operation": MOVE_FOLDER, + "album": proposal.album, + "source_path": str(source) if source else "", + "destination_path": str(destination) if destination else "", + "asset_ids": [asset.id for asset in members], + "expected_sha256": { + asset.id: asset.current_sha256 for asset in members if asset.current_sha256 + }, + "asset_count": len(members), + "same_filesystem": None, + "case_only": False, + "issues": issues, + } + + if source is None: + issues.append( + _issue("source_missing", f"no active assets remain in {proposal.album!r}") + ) + return operation + if not final_name: + issues.append(_issue("empty_name", f"proposal for {proposal.album!r} has no name")) + return operation + + self._validate_paths(operation, source, destination, issues) + self._validate_sources(operation, members, issues) + return operation + + def _source_folder(self, members: list[Asset]) -> Path | None: + if not members: + return None + return Path(members[0].current_path).parent + + def _validate_paths( + self, operation: dict, source: Path, destination: Path, issues: list[dict] + ) -> None: + if not self._roots: + issues.append(_issue("no_library_root", "no library root is configured")) + return + + for label, path in (("source", source), ("destination", destination)): + if path_policy.is_excluded(path): + issues.append(_issue("excluded_path", f"{label} is under an excluded directory")) + root = self._root_for(path) + if root is None: + issues.append( + _issue("root_escape", f"{label} {path} is outside every library root") + ) + + if source.is_symlink() or destination.is_symlink(): + issues.append(_issue("symlink", "refusing to rename through a symlink")) + + if not source.exists(): + issues.append(_issue("source_missing", f"source folder {source} does not exist")) + return + + # A symlinked source could still resolve outside the root even when its + # literal path looks fine; check the resolved location too. + root = self._root_for(source) + if root is not None: + try: + path_policy.resolve_within(root, source) + except path_policy.PathPolicyError as error: + issues.append(_issue("root_escape", str(error))) + + operation["case_only"] = ( + source != destination and source.as_posix().lower() == destination.as_posix().lower() + ) + operation["same_filesystem"] = self._same_filesystem(source, destination) + if operation["same_filesystem"] is False: + issues.append( + _issue( + "cross_filesystem", "destination is on another filesystem; copy-verify-delete" + ) + ) + + # A destination that already exists is only acceptable for a case-only + # rename, where source and destination are the same directory entry. + if destination.exists() and not operation["case_only"]: + issues.append(_issue("destination_exists", f"destination {destination} already exists")) + + def _validate_sources(self, operation: dict, members: list[Asset], issues: list[dict]) -> None: + """Every asset must still be present with the bytes inventory recorded.""" + for asset in members: + path = Path(asset.current_path) + if not path.exists(): + issues.append(_issue("source_missing", f"asset file {path} is missing")) + continue + expected = asset.byte_size + if expected is not None and path.stat().st_size != expected: + issues.append( + _issue("source_changed", f"asset file {path} changed since inventory") + ) + + def _root_for(self, path: Path) -> Path | None: + for root in self._roots: + resolved_root = Path(root) + if path == resolved_root or resolved_root in path.parents: + return resolved_root + return None + + def _same_filesystem(self, source: Path, destination: Path) -> bool | None: + anchor = destination.parent + if not source.exists() or not anchor.exists(): + return None + try: + return os.stat(source).st_dev == os.stat(anchor).st_dev + except OSError: + return None + + def _detect_cross_operation_issues(self, operations: list[dict]) -> None: + """Two operations must never target the same destination — including when the + names differ only by case or by Unicode normalization.""" + exact: dict[str, int] = {} + folded: dict[str, int] = {} + for operation in operations: + destination = operation["destination_path"] + if not destination: + continue + if destination in exact: + operation["issues"].append( + _issue("duplicate_target", f"destination {destination} is claimed twice") + ) + else: + exact[destination] = operation["sequence"] + + key = _nfc(destination).casefold() + if key in folded and folded[key] != operation["sequence"]: + operation["issues"].append( + _issue( + "unicode_collision", + f"destination {destination} collides with another target after " + "case/Unicode normalization", + ) + ) + else: + folded.setdefault(key, operation["sequence"]) + + def _persist(self, operations: list[dict]) -> dict: + blockers = sorted( + { + issue["code"] + for operation in operations + for issue in operation["issues"] + if issue["code"] in BLOCKING_CODES + } + ) + now = _now() + plan_id = str(uuid.uuid4()) + with self._session_factory() as session: + session.add( + RenamePlan( + id=plan_id, + state="invalid" if blockers else "validated", + schema_version=EXPORT_SCHEMA_VERSION, + checksum=plan_checksum(operations), + blockers=json.dumps(blockers), + operation_count=len(operations), + version=1, + validated_at=now, + ) + ) + # Flush the parent before its operations so the foreign key resolves. + session.flush() + for operation in operations: + session.add( + RenameOperation( + id=str(uuid.uuid4()), + plan_id=plan_id, + sequence=operation["sequence"], + operation=operation["operation"], + album=operation["album"], + source_path=operation["source_path"], + destination_path=operation["destination_path"], + asset_ids=json.dumps(operation["asset_ids"]), + expected_sha256=json.dumps(operation["expected_sha256"]), + asset_count=operation["asset_count"], + same_filesystem=operation["same_filesystem"], + case_only=operation["case_only"], + issues=json.dumps(operation["issues"], ensure_ascii=False), + ) + ) + session.commit() + return self.get(plan_id) + + # ── reads ───────────────────────────────────────────────────────────────── + + def get(self, plan_id: str) -> dict | None: + with self._session_factory() as session: + plan = session.get(RenamePlan, plan_id) + if plan is None: + return None + operations = list( + session.scalars( + select(RenameOperation) + .where(RenameOperation.plan_id == plan_id) + .order_by(RenameOperation.sequence) + ) + ) + return _plan_dict(plan, operations) + + def list_plans(self) -> dict: + with self._session_factory() as session: + plans = list(session.scalars(select(RenamePlan).order_by(RenamePlan.created_at.desc()))) + return { + "items": [ + { + "id": plan.id, + "state": plan.state, + "operation_count": plan.operation_count, + "blockers": json.loads(plan.blockers or "[]"), + "checksum": plan.checksum, + "version": plan.version, + } + for plan in plans + ], + "total": len(plans), + } + + def export(self, plan_id: str) -> dict: + """Portable JSON evidence for a plan. Contains no credentials.""" + plan = self.get(plan_id) + if plan is None: + raise RenameError(f"unknown rename plan {plan_id!r}") + return { + "schema_version": plan["schema_version"], + "plan_id": plan["id"], + "state": plan["state"], + "checksum": plan["checksum"], + "blockers": plan["blockers"], + "operations": [ + { + "sequence": operation["sequence"], + "operation": operation["operation"], + "album": operation["album"], + "source_path": operation["source_path"], + "destination_path": operation["destination_path"], + "asset_ids": operation["asset_ids"], + "asset_count": operation["asset_count"], + "case_only": operation["case_only"], + "same_filesystem": operation["same_filesystem"], + "issues": operation["issues"], + } + for operation in plan["operations"] + ], + } + + +def _plan_dict(plan: RenamePlan, operations: list[RenameOperation]) -> dict: + return { + "id": plan.id, + "state": plan.state, + "schema_version": plan.schema_version, + "checksum": plan.checksum, + "blockers": json.loads(plan.blockers or "[]"), + "operation_count": plan.operation_count, + "version": plan.version, + "applicable": plan.state == "validated", + "operations": [ + { + "sequence": operation.sequence, + "operation": operation.operation, + "album": operation.album, + "source_path": operation.source_path, + "destination_path": operation.destination_path, + "asset_ids": json.loads(operation.asset_ids or "[]"), + "expected_sha256": json.loads(operation.expected_sha256 or "{}"), + "asset_count": operation.asset_count, + "same_filesystem": operation.same_filesystem, + "case_only": operation.case_only, + "issues": json.loads(operation.issues or "[]"), + "journal_state": operation.journal_state, + } + for operation in operations + ], + } diff --git a/tests/integration/test_rename_plans.py b/tests/integration/test_rename_plans.py new file mode 100644 index 0000000..b2e1858 --- /dev/null +++ b/tests/integration/test_rename_plans.py @@ -0,0 +1,412 @@ +"""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 diff --git a/tests/story_traceability.json b/tests/story_traceability.json index 6d2a901..c9dedbd 100644 --- a/tests/story_traceability.json +++ b/tests/story_traceability.json @@ -84,6 +84,9 @@ ], "US03-05": [ "tests/e2e/test_phase_c_pipeline.py" + ], + "US04-01": [ + "tests/integration/test_rename_plans.py" ] } }