"""Offboarding completion freezes the leaver's evidence.

Three properties, each guarding a specific way this could quietly go
wrong:

* the archive actually happens (otherwise the whole feature is a no-op
  that nobody notices until an audit);
* completing twice does not stack duplicate snapshots an auditor would
  have to reconcile;
* an evidence failure never blocks the offboarding — the access removal
  is the security action, the paperwork is not.
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta

import pytest
from sqlalchemy import select

from app.audit.models import AuditLog
from app.core.access_cycle_listener import register_listener
from app.core.coordinators.offboarding_evidence_coordinator import (
    ARCHIVED_QUERIES,
    GENERATED_VIA_OFFBOARDING,
    archive_offboarding_evidence,
)
from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.auditor.models import AuditorEvidenceBundle

register_listener()


async def _seed_leaver_access(db, person_id):
    software = AssetSoftwareCatalog(name=f"OffSW {uuid.uuid4().hex[:6]}")
    db.add(software)
    await db.flush()
    plan = SoftwareLicensePlan(
        software_id=software.id, name="Pro", price=1, billing_cycle="monthly",
    )
    db.add(plan)
    await db.flush()
    db.add(SoftwareSeatAllocation(
        plan_id=plan.id, user_id=person_id, status="active",
        allocated_at=datetime.now(UTC) - timedelta(days=200),
    ))
    await db.flush()


async def _bundles(db, person_id) -> list[AuditorEvidenceBundle]:
    rows = (await db.execute(
        select(AuditorEvidenceBundle).where(
            AuditorEvidenceBundle.generated_via == GENERATED_VIA_OFFBOARDING,
        )
    )).scalars().all()
    return [b for b in rows if b.params.get("person_id") == str(person_id)]


@pytest.mark.asyncio
async def test_archive_freezes_both_leaver_questions(
    db_session, employee_user, admin_user,
):
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today()
    await db_session.flush()
    await _seed_leaver_access(db_session, employee.id)

    summary = await archive_offboarding_evidence(
        db_session,
        person_id=employee.id,
        process_id=uuid.uuid4(),
        trace_id=str(uuid.uuid4()),
        actor_user_id=admin.id,
    )
    assert {a["query_key"] for a in summary["archived"]} == set(ARCHIVED_QUERIES)
    assert not summary["failed"]

    stored = await _bundles(db_session, employee.id)
    assert len(stored) == len(ARCHIVED_QUERIES)
    assert all(b.content_hash for b in stored)
    # Person-scoped questions are not covered by the monthly company-wide
    # pre-compile, so this path is their only automatic producer.
    assert all(b.generated_via == GENERATED_VIA_OFFBOARDING for b in stored)


@pytest.mark.asyncio
async def test_completing_twice_does_not_stack_snapshots(
    db_session, employee_user, admin_user,
):
    """A retry or a re-run of the cron path must not leave an auditor
    reconciling two near-identical archives of the same departure."""
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today()
    await db_session.flush()
    await _seed_leaver_access(db_session, employee.id)

    # SAME process id both times — a retry, or a re-run of the cron path.
    process_id = uuid.uuid4()
    first = await archive_offboarding_evidence(
        db_session, person_id=employee.id, process_id=process_id,
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
    )
    second = await archive_offboarding_evidence(
        db_session, person_id=employee.id, process_id=process_id,
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
    )
    assert len(first["archived"]) == len(ARCHIVED_QUERIES)
    assert second["archived"] == []
    assert set(second["skipped"]) == set(ARCHIVED_QUERIES)
    assert len(await _bundles(db_session, employee.id)) == len(ARCHIVED_QUERIES)


@pytest.mark.asyncio
async def test_a_rehires_second_departure_is_archived_again(
    db_session, employee_user, admin_user,
):
    """Idempotency keys on the process AND its cycle.

    An earlier version of this test used two different ``process_id``
    values, which cannot happen: ``offboarding_processes.person_id`` is
    UNIQUE, so a boomerang rehire REOPENS the one row rather than
    inserting a second. Passing two ids therefore proved nothing about
    the real scenario, and the archive-side key could be deleted
    outright with the whole file still green. The same row with two
    ``leaver_date`` values is what production actually produces.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today()
    await db_session.flush()
    await _seed_leaver_access(db_session, employee.id)

    process_id = uuid.uuid4()  # ONE process, reopened.
    await archive_offboarding_evidence(
        db_session, person_id=employee.id, process_id=process_id,
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
        cycle="2024-03-31", anchor_at="2024-03-31",
    )
    summary = await archive_offboarding_evidence(
        db_session, person_id=employee.id, process_id=process_id,
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
        cycle="2026-08-14", anchor_at="2026-08-14",
    )
    assert len(summary["archived"]) == len(ARCHIVED_QUERIES)
    assert summary["skipped"] == []
    # Two departures, two sets of evidence.
    assert len(await _bundles(db_session, employee.id)) == 2 * len(ARCHIVED_QUERIES)


@pytest.mark.asyncio
async def test_the_same_departure_is_never_archived_twice(
    db_session, employee_user, admin_user,
):
    """The other half of the key: a retried completion must not stack.

    Same process, same cycle — a re-run of the completion path, which
    the runner does on retry — has to find the existing bundles and
    skip. Without the cycle in the comparison this passed for the wrong
    reason (process alone was enough); without the process it would pass
    for a different wrong reason. Both halves are load-bearing.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today()
    await db_session.flush()
    await _seed_leaver_access(db_session, employee.id)

    process_id = uuid.uuid4()
    for _ in range(2):
        summary = await archive_offboarding_evidence(
            db_session, person_id=employee.id, process_id=process_id,
            trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
            cycle="2026-08-14", anchor_at="2026-08-14",
        )
    assert summary["archived"] == []
    assert sorted(summary["skipped"]) == sorted(ARCHIVED_QUERIES)
    assert len(await _bundles(db_session, employee.id)) == len(ARCHIVED_QUERIES)


@pytest.mark.asyncio
async def test_an_unanchored_archive_declares_the_gap_in_the_bundle(
    db_session, employee_user, admin_user,
):
    """A degraded archive has to say so where the auditor reads it.

    Recording "no anchor" only in the coordinator's return value or an
    audit row moves the problem from silent to buried: the person
    opening the frozen bundle months later sees platform roles as of the
    run, with nothing marking them as such.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = None
    await db_session.flush()
    await _seed_leaver_access(db_session, employee.id)

    summary = await archive_offboarding_evidence(
        db_session, person_id=employee.id, process_id=uuid.uuid4(),
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
    )
    assert summary["anchor_gap"] == "no_termination_date"
    bundles = await _bundles(db_session, employee.id)
    footprint = next(b for b in bundles if b.query_key == "person_footprint")
    codes = {g.get("code") for g in (footprint.data_gaps or [])}
    assert "point_in_time_not_anchored" in codes


@pytest.mark.asyncio
async def test_a_second_leaver_is_archived_independently(
    db_session, employee_user, admin_user, manager_and_subordinate,
):
    """The idempotency check must key on the person, not on 'any archive
    exists' — otherwise the first leaver's archive suppresses everyone
    else's, and the gap is invisible."""
    employee, _ = employee_user
    admin, _ = admin_user
    _manager, other, _token = manager_and_subordinate

    for person in (employee, other):
        person.termination_date = date.today()
    await db_session.flush()

    await archive_offboarding_evidence(
        db_session, person_id=employee.id, process_id=uuid.uuid4(),
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
    )
    summary = await archive_offboarding_evidence(
        db_session, person_id=other.id, process_id=uuid.uuid4(),
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
    )
    assert len(summary["archived"]) == len(ARCHIVED_QUERIES)
    assert len(await _bundles(db_session, other.id)) == len(ARCHIVED_QUERIES)


@pytest.mark.asyncio
async def test_failure_is_audited_and_never_raises(
    db_session, db_engine, employee_user, admin_user, monkeypatch,
):
    """The access is already gone by the time this runs.

    Failing a real offboarding over an evidence query would trade a
    security action for a paperwork one.
    """
    from app.core.coordinators import offboarding_evidence_coordinator as coord

    employee, _ = employee_user
    admin, _ = admin_user
    process_id = uuid.uuid4()

    async def _boom(*_args, **_kwargs):
        raise RuntimeError("evidence source unavailable")

    monkeypatch.setattr(coord, "run_evidence_query", _boom, raising=False)
    monkeypatch.setattr(
        "app.domains.auditor.evidence_query.service.run_evidence_query",
        _boom,
    )
    # ``log_denied_action`` opens its own session through the module-level
    # factory (that independence is the point — the signal must outlive a
    # rollback). In tests that factory is not bound to the test database
    # unless the ``client`` fixture patched it, so bind it here or the
    # write goes nowhere and its exception is swallowed by design.
    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

    from app.core import database as _db

    monkeypatch.setattr(
        _db, "async_session_factory",
        async_sessionmaker(db_engine, class_=AsyncSession),
    )

    summary = await archive_offboarding_evidence(
        db_session, person_id=employee.id, process_id=process_id,
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
    )
    assert summary["archived"] == []
    assert len(summary["failed"]) == len(ARCHIVED_QUERIES)

    # ``log_denied_action`` writes in an INDEPENDENT session and commits
    # immediately, precisely so the signal survives a rollback of the
    # surrounding transaction. Reading it back therefore needs a separate
    # connection — this session's snapshot cannot see it, which is the
    # behaviour we want rather than a defect.
    factory = async_sessionmaker(db_engine, class_=AsyncSession)
    async with factory() as fresh:
        failures = (await fresh.execute(
            select(AuditLog).where(
                AuditLog.action == "offboarding.evidence_archive_failed",
                AuditLog.resource_id == str(process_id),
            )
        )).scalars().all()
    assert len(failures) == len(ARCHIVED_QUERIES), (
        "a swallowed failure must still leave a durable signal"
    )


@pytest.mark.asyncio
async def test_success_writes_one_archive_audit_row(
    db_session, employee_user, admin_user,
):
    employee, _ = employee_user
    admin, _ = admin_user
    process_id = uuid.uuid4()
    employee.termination_date = date.today()
    await db_session.flush()

    await archive_offboarding_evidence(
        db_session, person_id=employee.id, process_id=process_id,
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
    )
    row = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "offboarding.evidence_archived",
            AuditLog.resource_id == str(process_id),
        )
    )).scalar_one()
    assert row.detail["person_id"] == str(employee.id)
    assert len(row.detail["bundles"]) == len(ARCHIVED_QUERIES)
    # The hash is what makes the archive verifiable later.
    assert all(b["content_hash"] for b in row.detail["bundles"])


@pytest.mark.asyncio
async def test_archive_captures_still_held_access(
    db_session, employee_user, admin_user,
):
    """The point of archiving at completion.

    If the offboarding left something behind, the frozen bundle records
    it — rather than the gap being reconstructible only from tables that
    keep changing.
    """
    from app.domains.auditor.evidence_query.service import list_bundle_rows

    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today()
    await db_session.flush()
    await _seed_leaver_access(db_session, employee.id)  # never revoked

    await archive_offboarding_evidence(
        db_session, person_id=employee.id, process_id=uuid.uuid4(),
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
    )
    proof = next(
        b for b in await _bundles(db_session, employee.id)
        if b.query_key == "revocation_proof"
    )
    rows, _ = await list_bundle_rows(db_session, proof.id, page_size=200)
    assert any(r.payload.get("still_held") for r in rows)


@pytest.mark.asyncio
async def test_coordinator_does_not_import_across_domains():
    """The offboarding domain must not import the auditor domain.

    The seam is the reason this lives in ``core.coordinators``; a direct
    import would compile fine and quietly break the extraction boundary
    the platform is built on.
    """
    from pathlib import Path

    service = (
        Path(__file__).resolve().parents[2]
        / "app" / "domains" / "offboarding" / "service.py"
    ).read_text(encoding="utf-8")
    assert "offboarding_evidence_coordinator" in service
    assert "from app.domains.auditor" not in service


@pytest.mark.asyncio
async def test_footprint_is_anchored_to_the_last_day_not_to_now(
    db_session, employee_user, admin_user,
):
    """The archive runs AFTER the leaver's roles are revoked.

    An unanchored footprint therefore reports the departing admin
    holding no platform role at all — the single most misleading answer
    this archive could give, and precisely what an auditor opens it to
    check. The first attempt at this fix keyed on ``period_end``, which
    is auto-filled with RUN TIME when omitted, so it changed nothing on
    this path.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today() - timedelta(days=3)
    await db_session.flush()

    await archive_offboarding_evidence(
        db_session, person_id=employee.id, process_id=uuid.uuid4(),
        trace_id=str(uuid.uuid4()), actor_user_id=admin.id,
    )
    footprint = next(
        b for b in await _bundles(db_session, employee.id)
        if b.query_key == "person_footprint"
    )
    assert footprint.params.get("anchor_at") == (
        employee.termination_date.isoformat()
    ), "the point-in-time half must be anchored to the last day"