"""Access-drift detectors and their observe-only gate.

The gate is the part most worth testing. A detector let loose on years
of accumulated state can open a queue nobody reads, and a queue nobody
reads is worse than no queue — it trains people to ignore the one place
findings appear. So "observe unless explicitly told otherwise" has to be
true of the default, of an absent config key, and of a config key set to
anything other than an explicit ``false``.
"""
from __future__ import annotations

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

import pytest
from sqlalchemy import select

from app.auth.models import AuthRole, AuthUserRole, AuthUserRoleCycle
from app.core.access_cycle_listener import register_listener
from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.auditor.access_drift import (
    CONTROL_KEY,
    FINDING_DOUBLE_CYCLE,
    FINDING_LEAVER_ACCESS,
    FINDING_ROLE_WITHOUT_LEDGER,
    SETTING_OFFBOARDING_ARCHIVE_EPOCH,
    SETTING_ROLE_LEDGER_EPOCH,
    detect_double_open_cycles,
    detect_leaver_access,
    detect_roles_without_ledger,
    run_access_drift,
)
from app.domains.auditor.models import AuditorFinding

register_listener()


async def _mk_leaver(db, *, days_ago: int = 30):
    """A person who is ALREADY a leaver.

    Deliberately not ``person.employment_status = "left"`` on an existing
    row: the Leaver Blocked listener refuses that flip outright and
    requires ``LeaverBlockGuard.flip_to_left()``. The first version of
    these tests did exactly that and passed in isolation — the guard is
    only registered once some other test pulls in the ``client`` fixture,
    so the file alone was green while the whole suite failed.
    """
    from app.identity.person_models import IdentityPerson

    person = IdentityPerson(
        work_email=f"leaver-{uuid.uuid4().hex[:8]}@test.com",
        display_name="Drift Leaver",
        employment_status="left",
        termination_date=date.today() - timedelta(days=days_ago),
        hris_person_id=f"DRIFT{uuid.uuid4().hex[:6]}",
    )
    db.add(person)
    await db.flush()
    return person


async def _seed_seat(db, person_id, *, revoked=False):
    software = AssetSoftwareCatalog(name=f"DriftSW {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="revoked" if revoked else "active",
        allocated_at=datetime.now(UTC) - timedelta(days=200),
        revoked_at=datetime.now(UTC) if revoked else None,
    ))
    await db.flush()


# ── observe-only gate ─────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_observe_mode_opens_no_findings(db_session, employee_user):
    employee = await _mk_leaver(db_session, days_ago=30)
    await _seed_seat(db_session, employee.id)

    summary = await run_access_drift(db_session, observe_only=True)
    assert summary["observe_only"] is True
    assert summary["would_open_total"] > 0, "the drift is real"

    opened = (await db_session.execute(
        select(AuditorFinding).where(AuditorFinding.control_key == CONTROL_KEY)
    )).scalars().all()
    assert opened == [], "observe mode must not create tickets"


@pytest.mark.asyncio
async def test_enforce_mode_opens_findings(db_session, employee_user):
    employee = await _mk_leaver(db_session, days_ago=30)
    await _seed_seat(db_session, employee.id)

    summary = await run_access_drift(db_session, observe_only=False)
    assert summary["observe_only"] is False
    opened = (await db_session.execute(
        select(AuditorFinding).where(
            AuditorFinding.control_key == CONTROL_KEY,
            AuditorFinding.finding_type == FINDING_LEAVER_ACCESS,
        )
    )).scalars().all()
    assert opened, "enforce mode must actually open the ticket"


@pytest.mark.asyncio
async def test_absent_or_null_config_never_opens_tickets():
    """An absent key meaning "open everything" is how a first deploy
    turns into a ticket storm.

    The stricter half is that an UNRECOGNISED value raises rather than
    falling back. The original ``is not False`` check silently kept
    observe mode for every spelling except the JSON literal, so an
    operator who wrote ``"observe_only": "false"`` believed they had
    enabled ticketing and had not — and nothing anywhere said so.
    """
    from app.scheduled_jobs.access_drift_handler import run_access_drift_scan

    seen: list[bool] = []

    async def _capture(_db, *, observe_only, grace_days=None):
        seen.append(observe_only)
        return {"observe_only": observe_only, "by_type": {}}

    import app.domains.auditor.access_drift as drift_mod

    async def _run(config):
        real = drift_mod.run_access_drift
        drift_mod.run_access_drift = _capture
        try:
            return await run_access_drift_scan(None, config)
        finally:
            drift_mod.run_access_drift = real

    for config in ({}, {"observe_only": None}, {"observe_only": True},
                   {"observe_only": "true"}, {"observe_only": 1}):
        await _run(config)
    assert seen == [True] * 5, "observe mode is the default and stays it"

    # Every spelling an operator would plausibly write must flip it.
    for config in ({"observe_only": False}, {"observe_only": "false"},
                   {"observe_only": 0}, {"observe_only": "no"}):
        await _run(config)
    assert seen[-4:] == [False] * 4

    with pytest.raises(RuntimeError, match="observe_only"):
        await _run({"observe_only": "maybe"})


@pytest.mark.asyncio
async def test_a_malformed_grace_window_raises_instead_of_defaulting():
    """Silently falling back left an operator who wrote "7 days"
    believing they had widened the window."""
    from app.scheduled_jobs.access_drift_handler import run_access_drift_scan

    with pytest.raises(RuntimeError, match="grace_days"):
        await run_access_drift_scan(None, {"grace_days": "7 days"})


# ── leaver access ─────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_leaver_still_holding_a_seat_is_detected(
    db_session, employee_user,
):
    employee = await _mk_leaver(db_session, days_ago=30)
    await _seed_seat(db_session, employee.id)

    specs = await detect_leaver_access(db_session)
    kinds = {s.detail["access_kind"] for s in specs}
    assert "software seat" in kinds
    assert all(s.severity == "high" for s in specs)


@pytest.mark.asyncio
async def test_revoked_access_is_not_drift(db_session, employee_user):
    employee = await _mk_leaver(db_session, days_ago=30)
    await _seed_seat(db_session, employee.id, revoked=True)

    specs = await detect_leaver_access(db_session)
    assert not [s for s in specs if s.detail["access_kind"] == "software seat"]


@pytest.mark.asyncio
async def test_a_fresh_leaver_is_inside_the_grace_window(
    db_session, employee_user,
):
    """Access is not drift the instant someone leaves — provisioning and
    the offboarding queue need a window. Alerting inside it would report
    every well-run offboarding as a failure."""
    employee = await _mk_leaver(db_session, days_ago=1)
    await _seed_seat(db_session, employee.id)

    assert await detect_leaver_access(db_session) == []
    # Past the window the same state IS a finding.
    assert await detect_leaver_access(db_session, grace_days=0)


@pytest.mark.asyncio
async def test_active_employees_are_never_flagged(db_session, employee_user):
    employee, _ = employee_user
    await _seed_seat(db_session, employee.id)
    assert await detect_leaver_access(db_session) == []


# ── ledger integrity ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_double_open_cycle_is_detected(db_session, employee_user):
    """The ledgers' one-open-cycle invariant is deliberately NOT a UNIQUE
    constraint — an evidence ledger must never be able to fail the
    business transaction it observes — so it is enforced here, where a
    violation becomes a ticket instead of a 500 on someone's grant."""
    employee, _ = employee_user
    role_id = (await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "auditor")
    )).scalar_one()

    now = datetime.now(UTC)
    for offset in (0, 1):
        db_session.add(AuthUserRoleCycle(
            user_id=employee.id,
            role_id=role_id,
            role_name="auditor",
            granted_at=now - timedelta(seconds=offset),
            grant_source="manual",
        ))
    await db_session.flush()

    specs = await detect_double_open_cycles(db_session)
    assert any(s.detail.get("open_cycles", 0) >= 2 for s in specs)
    assert all(s.severity == "medium" for s in specs)


@pytest.mark.asyncio
async def test_a_single_open_cycle_is_not_drift(db_session, employee_user):
    employee, _ = employee_user
    role_id = (await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "auditor")
    )).scalar_one()
    db_session.add(AuthUserRole(user_id=employee.id, role_id=role_id))
    await db_session.flush()

    specs = await detect_double_open_cycles(db_session)
    assert not [
        s for s in specs
        if s.detail.get("ledger") == "auth_user_role_cycles"
        and s.subject_ref == str(employee.id)
    ]


async def _set_epoch(db, key: str, when: datetime) -> None:
    """Declare when a control became measurable.

    Every drift test that expects a finding has to say this, because a
    detector with no epoch judges nothing on purpose — see migration 511
    and the 617 false positives that motivated it.
    """
    from app.integrations.models import PlatformSetting

    db.add(PlatformSetting(key=key, value=when.isoformat()))
    await db.flush()


@pytest.mark.asyncio
async def test_a_control_with_no_epoch_judges_nothing(
    db_session, employee_user,
):
    """The 617-false-positive guard.

    On production data this detector would have opened 617 findings, and
    its own docstring already said why: a grant predating migration 484
    has no interval BY DESIGN, because 484 deliberately does not backfill.
    With no recorded start date a detector has no basis for calling
    anything drift, so it must judge nothing rather than everything.
    """
    from sqlalchemy import insert

    employee, _ = employee_user
    role_id = (await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "finance")
    )).scalar_one()
    await db_session.execute(
        insert(AuthUserRole).values(
            user_id=employee.id, role_id=role_id, grant_source="manual",
        )
    )
    await db_session.flush()

    assert await detect_roles_without_ledger(db_session) == []


@pytest.mark.asyncio
async def test_a_grant_predating_the_ledger_is_not_drift(
    db_session, employee_user,
):
    """Migration 484 does not backfill, so a grant made before the ledger
    existed cannot have an interval. Flagging it is a false positive by
    construction, not a finding."""
    from sqlalchemy import insert

    employee, _ = employee_user
    role_id = (await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "finance")
    )).scalar_one()
    now = datetime.now(UTC)
    await db_session.execute(
        insert(AuthUserRole).values(
            user_id=employee.id, role_id=role_id, grant_source="manual",
            granted_at=now - timedelta(days=120),
        )
    )
    await _set_epoch(db_session, SETTING_ROLE_LEDGER_EPOCH, now - timedelta(days=1))

    assert await detect_roles_without_ledger(db_session) == []


@pytest.mark.asyncio
async def test_the_employee_role_is_never_privileged_drift(
    db_session, employee_user,
):
    """566 of the 617 were somebody's plain ``employee`` role, which
    confers self scope only. A finding named *privileged*_role_... whose
    query means "everyone" trains people to close the queue unread."""
    from sqlalchemy import delete, insert

    employee, _ = employee_user
    now = datetime.now(UTC)
    role_ids = {
        name: rid for name, rid in (await db_session.execute(
            select(AuthRole.name, AuthRole.id)
            .where(AuthRole.name.in_(("employee", "finance")))
        )).all()
    }
    # Re-create the employee holding OUTSIDE the listener so it has no
    # ledger interval — otherwise the ledger check excludes it and this
    # test passes whether the privilege filter exists or not. (It did:
    # putting ``employee`` back into the privileged set left the suite
    # green until this was rewritten.)
    await db_session.execute(
        delete(AuthUserRole).where(
            AuthUserRole.user_id == employee.id,
            AuthUserRole.role_id == role_ids["employee"],
        )
    )
    await db_session.flush()
    await db_session.execute(
        delete(AuthUserRoleCycle).where(
            AuthUserRoleCycle.user_id == employee.id,
        )
    )
    for name in ("employee", "finance"):
        await db_session.execute(
            insert(AuthUserRole).values(
                user_id=employee.id, role_id=role_ids[name],
                grant_source="manual", granted_at=now,
            )
        )
    await db_session.flush()
    await _set_epoch(db_session, SETTING_ROLE_LEDGER_EPOCH, now - timedelta(days=1))

    flagged = {
        spec.detail["role"] for spec in
        await detect_roles_without_ledger(db_session)
        if spec.subject_ref == str(employee.id)
    }
    # Same person, same instant, same missing interval — only privilege
    # separates the two, so ``finance`` proves the detector is working
    # while ``employee`` proves the filter is what excluded it.
    assert "finance" in flagged
    assert "employee" not in flagged


@pytest.mark.asyncio
async def test_role_written_outside_the_orm_is_detected(
    db_session, employee_user,
):
    """A role the listener never saw is invisible to every point-in-time
    query — which is the exact failure this whole feature exists to stop.

    Simulated with a Core INSERT, the one write shape the ORM listener
    genuinely cannot observe.
    """
    from sqlalchemy import insert

    employee, _ = employee_user
    role_id = (await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "finance")
    )).scalar_one()
    await db_session.execute(
        insert(AuthUserRole).values(
            user_id=employee.id, role_id=role_id, grant_source="manual",
            granted_at=datetime.now(UTC),
        )
    )
    await _set_epoch(
        db_session, SETTING_ROLE_LEDGER_EPOCH,
        datetime.now(UTC) - timedelta(days=1),
    )

    specs = await detect_roles_without_ledger(db_session)
    assert any(
        s.detail["role"] == "finance" and s.subject_ref == str(employee.id)
        for s in specs
    ), "a Core-level grant leaves no ledger interval and must be flagged"


@pytest.mark.asyncio
async def test_a_normally_granted_role_is_not_flagged(
    db_session, employee_user,
):
    employee, _ = employee_user
    role_id = (await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "finance")
    )).scalar_one()
    db_session.add(AuthUserRole(user_id=employee.id, role_id=role_id))
    await db_session.flush()
    await _set_epoch(
        db_session, SETTING_ROLE_LEDGER_EPOCH,
        datetime.now(UTC) - timedelta(days=1),
    )

    specs = await detect_roles_without_ledger(db_session)
    assert not [
        s for s in specs
        if s.detail["role"] == "finance" and s.subject_ref == str(employee.id)
    ]


# ── findings lifecycle ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_resolved_drift_auto_closes_the_finding(
    db_session, employee_user,
):
    """A finding whose condition is gone must close itself, or the queue
    fills with work that is already done."""
    employee = await _mk_leaver(db_session, days_ago=30)
    await _seed_seat(db_session, employee.id)

    await run_access_drift(db_session, observe_only=False)
    seat = (await db_session.execute(
        select(SoftwareSeatAllocation).where(
            SoftwareSeatAllocation.user_id == employee.id,
        )
    )).scalars().first()
    seat.revoked_at = datetime.now(UTC)
    seat.status = "revoked"
    await db_session.flush()

    summary = await run_access_drift(db_session, observe_only=False)
    assert summary["by_type"][FINDING_LEAVER_ACCESS]["reconciled"][
        "auto_resolved"
    ] >= 1


@pytest.mark.asyncio
async def test_rerunning_does_not_duplicate_findings(
    db_session, employee_user,
):
    employee = await _mk_leaver(db_session, days_ago=30)
    await _seed_seat(db_session, employee.id)

    await run_access_drift(db_session, observe_only=False)
    await run_access_drift(db_session, observe_only=False)

    rows = (await db_session.execute(
        select(AuditorFinding).where(
            AuditorFinding.control_key == CONTROL_KEY,
            AuditorFinding.finding_type == FINDING_LEAVER_ACCESS,
        )
    )).scalars().all()
    assert len({r.dedup_key for r in rows}) == len(rows)


@pytest.mark.asyncio
async def test_every_detector_reports_a_count_in_observe_mode(db_session):
    """The whole point of observe mode is measuring the first-day volume
    before anything is opened."""
    summary = await run_access_drift(db_session, observe_only=True)
    from app.domains.auditor.access_drift import DETECTORS

    assert set(summary["by_type"]) == {name for name, _ in DETECTORS}
    assert all("detected" in v for v in summary["by_type"].values())
    assert all("reconciled" not in v for v in summary["by_type"].values())


@pytest.mark.asyncio
async def test_job_registered_in_both_processes():
    from pathlib import Path

    from app.scheduled_jobs.registry import get_handler

    assert get_handler("auditor.access_drift_scan") is not None
    root = Path(__file__).resolve().parents[2] / "app"
    for process in ("main.py", "worker.py"):
        assert "access_drift_handler" in (
            root / process
        ).read_text(encoding="utf-8"), process


@pytest.mark.asyncio
async def test_finding_types_are_distinct():
    """Two detectors sharing a finding_type would auto-resolve each
    other's tickets on every run."""
    from app.domains.auditor.access_drift import DETECTORS

    names = [name for name, _ in DETECTORS]
    assert len(names) == len(set(names))
    assert FINDING_DOUBLE_CYCLE in names
    assert FINDING_ROLE_WITHOUT_LEDGER in names


# ── missing-evidence detector: the legacy fallback's boundaries ───────


async def _mk_offboarding(db, person_id, *, completed_days_ago: int, proc=None):
    """Complete an offboarding, reopening an existing process if given.

    Reopening rather than inserting is what production does: ``person_id``
    is UNIQUE on ``offboarding_processes``, so a boomerang rehire resets
    the same row (status, ``leaver_date``, ``completed_at``) instead of
    creating a second one.
    """
    from app.domains.offboarding.models import OffboardingProcess

    completed = datetime.now(UTC) - timedelta(days=completed_days_ago)
    if proc is None:
        proc = OffboardingProcess(person_id=person_id, leaver_date=completed.date())
        db.add(proc)
    proc.leaver_date = completed.date()
    proc.status = "completed"
    proc.completed_at = completed
    await db.flush()
    return proc, completed


async def _mk_bundle(db, *, query_key, params, created_at):
    """A bundle written by the offboarding archive path."""
    from app.core.coordinators.offboarding_evidence_coordinator import (
        GENERATED_VIA_OFFBOARDING,
    )
    from app.domains.auditor.models import AuditorEvidenceBundle

    bundle = AuditorEvidenceBundle(
        query_key=query_key,
        query_version="v1",
        params=params,
        row_count=1,
        content_hash=uuid.uuid4().hex,
        generated_via=GENERATED_VIA_OFFBOARDING,
        created_at=created_at,
    )
    db.add(bundle)
    await db.flush()
    return bundle


@pytest.mark.asyncio
async def test_a_rehires_second_departure_is_not_covered_by_the_first(
    db_session, employee_user,
):
    """The regression the process key was introduced to prevent.

    Someone leaves, is rehired, and leaves again. The first departure's
    person-keyed archive must not answer for the second — matching on
    the person alone made a genuinely unarchived offboarding invisible,
    which is the one thing this detector exists to catch.
    """
    from app.core.coordinators.offboarding_evidence_coordinator import (
        ARCHIVED_QUERIES,
    )
    from app.domains.auditor.access_drift import (
        detect_missing_offboarding_evidence,
    )

    person = await _mk_leaver(db_session, days_ago=3)
    proc, first_exit = await _mk_offboarding(
        db_session, person.id, completed_days_ago=400,
    )
    # The first departure was archived — under BOTH the old person key
    # and the process key, because the process row is the same row the
    # second departure will reuse.
    for query_key in ARCHIVED_QUERIES:
        await _mk_bundle(
            db_session,
            query_key=query_key,
            params={"person_id": str(person.id)},
            created_at=first_exit,
        )
        await _mk_bundle(
            db_session,
            query_key=query_key,
            params={
                "person_id": str(person.id),
                "offboarding_process_id": str(proc.id),
                "offboarding_cycle": proc.leaver_date.isoformat(),
            },
            created_at=first_exit,
        )
    new_proc, _ = await _mk_offboarding(
        db_session, person.id, completed_days_ago=3, proc=proc,
    )

    await _set_epoch(
        db_session, SETTING_OFFBOARDING_ARCHIVE_EPOCH,
        datetime.now(UTC) - timedelta(days=365),
    )
    specs = await detect_missing_offboarding_evidence(db_session)

    mine = [s for s in specs if s.subject_ref == str(new_proc.id)]
    assert mine, "second departure has no archive and must be flagged"
    assert set(mine[0].detail["missing_queries"]) == set(ARCHIVED_QUERIES)


@pytest.mark.asyncio
async def test_a_leaver_holding_a_privileged_role_is_found_without_any_ledger(
    db_session,
):
    """The false NEGATIVE, which mattered more than the false positives.

    This control read ``auth_user_role_cycles`` — the ledger — to answer
    "does this leaver still hold a platform role". Migration 484
    deliberately does not backfill, so on production every holding
    predates the ledger and the query matched nothing: 120 departed people
    held a role the detector could not see. All 120 were plain
    ``employee`` (harmless, and excluded), so nothing serious was missed —
    but the day somebody leaves holding ``global_admin``, silence is the
    wrong answer. The live table is the only authority on access NOW.
    """
    from sqlalchemy import insert

    leaver = await _mk_leaver(db_session, days_ago=30)
    role_id = (await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "global_admin")
    )).scalar_one()
    # Core INSERT so the listener never fires: no ledger interval exists,
    # which is exactly the production shape.
    await db_session.execute(
        insert(AuthUserRole).values(
            user_id=leaver.id, role_id=role_id, grant_source="manual",
            granted_at=datetime.now(UTC) - timedelta(days=200),
        )
    )
    await db_session.flush()
    assert not (await db_session.execute(
        select(AuthUserRoleCycle.id)
        .where(AuthUserRoleCycle.user_id == leaver.id)
    )).scalars().all(), "this test is only meaningful with an empty ledger"

    specs = await detect_leaver_access(db_session)
    assert any(
        spec.detail.get("access_kind") == "platform role"
        and spec.detail.get("access_label") == "global_admin"
        and spec.subject_ref == str(leaver.id)
        for spec in specs
    ), "a leaver still holding global_admin must be found from the live table"


@pytest.mark.asyncio
async def test_a_leavers_employee_role_is_not_reported_as_held_access(
    db_session,
):
    """``employee`` is self scope only and harmless on a disabled account —
    which is why rbac.py excludes it from ``LEAVER_REVOCABLE_ROLES``.
    Reading the live table must not turn 120 departed people into 120
    tickets."""
    from sqlalchemy import insert

    leaver = await _mk_leaver(db_session, days_ago=30)
    role_id = (await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "employee")
    )).scalar_one()
    await db_session.execute(
        insert(AuthUserRole).values(
            user_id=leaver.id, role_id=role_id, grant_source="manual",
            granted_at=datetime.now(UTC) - timedelta(days=200),
        )
    )
    await db_session.flush()

    specs = await detect_leaver_access(db_session)
    assert not [
        spec for spec in specs
        if spec.detail.get("access_kind") == "platform role"
        and spec.subject_ref == str(leaver.id)
    ], "employee on a departed account is not access worth a ticket"