photoanalyzer/tests/integration/test_rename_recovery.py

"""Recovering or rolling back interrupted renames (US04-04).

The crash tests are real: a child process applies a plan and is killed by the
``PHOTO_PIPELINE_FAULT_AFTER`` barrier at each persisted journal transition in turn.
The parent then reopens the database, classifies the wreckage from journal + disk
evidence, and asserts recovery converges without losing or overwriting anything.
"""

import os
import subprocess
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path

import pytest
from sqlalchemy import select

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 import hashing
from photo_pipeline.services.rename_apply import ApplyError, RenameApplyService
from photo_pipeline.services.rename_journal import (
    MANUAL,
    RESUMABLE,
    ROLLBACK_SAFE,
    JournalState,
)
from photo_pipeline.services.renames import RenameService

REPO = Path(__file__).resolve().parents[2]
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)

# A child process that applies the plan and dies at the configured barrier.
APPLY_SCRIPT = """
import sys
sys.path.insert(0, {repo!r})
from photo_pipeline.db import create_db_engine, create_session_factory
from photo_pipeline.services.rename_apply import RenameApplyService

db_url, lib, plan_id, version = sys.argv[1], sys.argv[2], sys.argv[3], int(sys.argv[4])
sf = create_session_factory(create_db_engine(db_url))
RenameApplyService(sf, library_roots=(lib,)).apply(plan_id, expected_version=version)
"""


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",)):
    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 _crash_during_apply(config, lib, plan, barrier, tmp_path):
    """Run apply in a child that dies right after ``barrier`` is persisted."""
    script = tmp_path / f"apply_{barrier}.py"
    script.write_text(APPLY_SCRIPT.format(repo=str(REPO)))
    env = dict(os.environ)
    env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier
    result = subprocess.run(
        [
            sys.executable,
            str(script),
            config.database_url,
            str(lib),
            plan["id"],
            str(plan["version"]),
        ],
        env=env,
        capture_output=True,
    )
    assert result.returncode == -9 or result.returncode == 9, (
        f"child should have been killed at {barrier}, got {result.returncode}: "
        f"{result.stderr.decode(errors='replace')[-400:]}"
    )


def _assets(sf):
    with sf() as session:
        return {a.id: a.current_path for a in session.scalars(select(Asset))}


def _files(lib):
    return {p.read_bytes() for p in lib.rglob("*") if p.is_file() and ".rename-" not in str(p)}


# ── fault injection at every persisted transition ────────────────────────────


@pytest.mark.parametrize(
    "barrier",
    [
        JournalState.MOVING,
        JournalState.MOVED,
        JournalState.DATABASE_UPDATED,
        JournalState.VERIFIED,
    ],
)
def test_crash_at_each_transition_recovers_without_losing_content(barrier, tmp_path):
    config, sf, lib = _factory(tmp_path)
    _album(sf, lib, "rome", approved_name="2019 Rome")
    plan = _plan(sf, lib)
    content_before = _files(lib)

    _crash_during_apply(config, lib, plan, barrier, tmp_path)

    # Fresh process: the journal plus the disk decide what happened.
    reopened = create_session_factory(create_db_engine(config.database_url))
    service = RenameApplyService(reopened, library_roots=(lib,))
    verdicts = service.journal.classify_all()
    assert verdicts, f"a crash at {barrier} must leave recoverable work"
    assert all(v["classification"] in {RESUMABLE, ROLLBACK_SAFE} for v in verdicts), verdicts

    service.recover()

    # No content was lost anywhere in the library.
    assert _files(lib) == content_before
    # And nothing is left half-done.
    assert service.journal.blocks_mutation() is False


@pytest.mark.parametrize(
    "barrier", [JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED]
)
def test_recovery_is_idempotent_across_repeated_restarts(barrier, tmp_path):
    config, sf, lib = _factory(tmp_path)
    _album(sf, lib, "rome", approved_name="2019 Rome")
    plan = _plan(sf, lib)
    _crash_during_apply(config, lib, plan, barrier, tmp_path)

    reopened = create_session_factory(create_db_engine(config.database_url))
    service = RenameApplyService(reopened, library_roots=(lib,))
    service.recover()
    settled_paths = _assets(reopened)
    settled_files = _files(lib)

    # Repeated recovery passes converge — the second and third change nothing.
    for _ in range(2):
        service.recover()
        assert _assets(reopened) == settled_paths
        assert _files(lib) == settled_files


def test_crash_before_the_move_leaves_the_source_intact_and_resumable(tmp_path):
    config, sf, lib = _factory(tmp_path)
    _album(sf, lib, "rome", approved_name="2019 Rome")
    plan = _plan(sf, lib)
    _crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path)

    reopened = create_session_factory(create_db_engine(config.database_url))
    service = RenameApplyService(reopened, library_roots=(lib,))
    verdict = service.journal.classify_all()[0]
    assert verdict["classification"] == RESUMABLE
    assert (lib / "rome").exists() and not (lib / "2019 Rome").exists()

    service.recover()
    # Resumable work returns to `planned`, so a fresh apply can finish the job.
    operation = service.journal.operations(plan["id"])[0]
    assert operation["journal_state"] == JournalState.PLANNED

    fresh = RenameService(reopened, library_roots=(lib,)).get(plan["id"])
    service.apply(plan["id"], expected_version=fresh["version"])
    assert (lib / "2019 Rome").is_dir() and not (lib / "rome").exists()


def test_crash_after_the_move_completes_the_bookkeeping(tmp_path):
    config, sf, lib = _factory(tmp_path)
    _album(sf, lib, "rome", approved_name="2019 Rome")
    plan = _plan(sf, lib)
    _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)

    reopened = create_session_factory(create_db_engine(config.database_url))
    service = RenameApplyService(reopened, library_roots=(lib,))
    assert service.journal.classify_all()[0]["classification"] == ROLLBACK_SAFE

    service.recover()
    operation = service.journal.operations(plan["id"])[0]
    assert operation["journal_state"] == JournalState.COMPLETE
    # The database caught up with the filesystem.
    assert all("2019 Rome" in path for path in _assets(reopened).values())


# ── ambiguous evidence ───────────────────────────────────────────────────────


def test_an_unexpected_destination_occupant_forces_manual_recovery(tmp_path):
    config, sf, lib = _factory(tmp_path)
    _album(sf, lib, "rome", approved_name="2019 Rome")
    plan = _plan(sf, lib)
    _crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path)

    # Someone puts something at the destination while the app was down.
    (lib / "2019 Rome").mkdir()
    (lib / "2019 Rome" / "stranger.jpg").write_bytes(b"not ours")

    reopened = create_session_factory(create_db_engine(config.database_url))
    service = RenameApplyService(reopened, library_roots=(lib,))
    verdict = service.journal.classify_all()[0]
    assert verdict["classification"] == MANUAL
    assert verdict["reason"]

    result = service.recover()
    assert result["manual"] == 1
    # Nothing was touched, and the ambiguity keeps blocking other mutations.
    assert (lib / "2019 Rome" / "stranger.jpg").read_bytes() == b"not ours"
    assert (lib / "rome").exists()
    assert service.journal.blocks_mutation() is True


def test_manual_work_blocks_a_different_plan_from_applying(tmp_path):
    config, sf, lib = _factory(tmp_path)
    _album(sf, lib, "rome", approved_name="2019 Rome")
    plan = _plan(sf, lib)
    _crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path)
    (lib / "2019 Rome").mkdir()  # ambiguous

    reopened = create_session_factory(create_db_engine(config.database_url))
    # The user parks the stuck album (un-approves it) and tries a different one, so
    # the new plan is valid on its own — only the unresolved journal entry stands in
    # the way.
    with reopened() as session:
        stuck = session.scalar(select(AlbumProposal).where(AlbumProposal.album == "rome"))
        stuck.status = "edited"
        session.commit()
    _album(reopened, lib, "paris", approved_name="2020 Paris")
    other = RenameService(reopened, library_roots=(lib,)).build_plan()
    assert other["state"] == "validated", "the unrelated plan itself must be applyable"

    service = RenameApplyService(reopened, library_roots=(lib,))
    with pytest.raises(ApplyError, match="another rename is unresolved"):
        service.apply(other["id"], expected_version=other["version"])
    # Nothing moved for the unrelated album either.
    assert (lib / "paris").exists() and not (lib / "2020 Paris").exists()


# ── rollback ─────────────────────────────────────────────────────────────────


def test_rollback_returns_the_content_to_its_source(tmp_path):
    config, sf, lib = _factory(tmp_path)
    _album(sf, lib, "rome", approved_name="2019 Rome")
    plan = _plan(sf, lib)
    before = _files(lib)
    _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)

    reopened = create_session_factory(create_db_engine(config.database_url))
    service = RenameApplyService(reopened, library_roots=(lib,))
    operation = service.journal.operations(plan["id"])[0]

    service.rollback_operation(operation["id"])

    assert (lib / "rome").is_dir() and not (lib / "2019 Rome").exists()
    assert _files(lib) == before
    assert service.journal.get(operation["id"])["journal_state"] == JournalState.ROLLED_BACK
    assert all("rome/" in path for path in _assets(reopened).values())


def test_rollback_refuses_when_the_source_is_occupied(tmp_path):
    config, sf, lib = _factory(tmp_path)
    _album(sf, lib, "rome", approved_name="2019 Rome")
    plan = _plan(sf, lib)
    _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)

    # Something new appears at the old location while the app was down.
    (lib / "rome").mkdir()
    (lib / "rome" / "new.jpg").write_bytes(b"newcomer")

    reopened = create_session_factory(create_db_engine(config.database_url))
    service = RenameApplyService(reopened, library_roots=(lib,))
    operation = service.journal.operations(plan["id"])[0]

    with pytest.raises(ApplyError, match="occupied"):
        service.rollback_operation(operation["id"])
    assert (lib / "rome" / "new.jpg").read_bytes() == b"newcomer"


def test_rollback_refuses_content_whose_bytes_changed(tmp_path):
    config, sf, lib = _factory(tmp_path)
    _album(sf, lib, "rome", approved_name="2019 Rome")
    plan = _plan(sf, lib)
    _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)

    # The moved content is edited before rollback is attempted.
    (lib / "2019 Rome" / "a.jpg").write_bytes(b"edited after the move")

    reopened = create_session_factory(create_db_engine(config.database_url))
    service = RenameApplyService(reopened, library_roots=(lib,))
    operation = service.journal.operations(plan["id"])[0]

    with pytest.raises(ApplyError, match="manual recovery"):
        service.rollback_operation(operation["id"])
    assert (lib / "2019 Rome" / "a.jpg").read_bytes() == b"edited after the move"


def test_rollback_is_idempotent(tmp_path):
    config, sf, lib = _factory(tmp_path)
    _album(sf, lib, "rome", approved_name="2019 Rome")
    plan = _plan(sf, lib)
    _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)

    reopened = create_session_factory(create_db_engine(config.database_url))
    service = RenameApplyService(reopened, library_roots=(lib,))
    operation = service.journal.operations(plan["id"])[0]
    service.rollback_operation(operation["id"])
    settled = _files(lib)

    # A second rollback has nothing left to do and refuses rather than moving again.
    with pytest.raises(ApplyError):
        service.rollback_operation(operation["id"])
    assert _files(lib) == settled