"""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