"""Person footprint, revocation proof, object lifecycle, review proof.

The two assertions worth stating up front, because both encode a way an
evidence tool can lie by omission:

* **Revocation proof must include what was NOT revoked.** A query that
  returns only successful revocations answers "all clear" by
  construction — the exact false assurance an audit exists to catch.
* **Review proof must include undecided items.** A campaign reporting
  only its completed items always looks complete; the overdue tail is
  the finding.
"""
from __future__ import annotations

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

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.models import AuditLog
from app.core.access_cycle_listener import register_listener
from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.auditor.evidence_query.service import (
    list_bundle_rows,
    run_evidence_query,
)
from app.domains.auditor.models import (
    AccessCertification,
    AccessCertificationItem,
)
from app.domains.cmdb.models import CmdbNetworkOwnerHistory

register_listener()


async def _payloads(db: AsyncSession, bundle_id) -> list[dict]:
    rows, _ = await list_bundle_rows(db, bundle_id, page_size=200)
    return [r.payload for r in rows]


async def _mk_seat(
    db: AsyncSession, person_id, *, allocated_at, revoked_at=None,
) -> SoftwareSeatAllocation:
    software = AssetSoftwareCatalog(name=f"LC SW {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()
    seat = SoftwareSeatAllocation(
        plan_id=plan.id,
        user_id=person_id,
        status="revoked" if revoked_at else "active",
        allocated_at=allocated_at,
        revoked_at=revoked_at,
    )
    db.add(seat)
    await db.flush()
    return seat


# ── person footprint ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_footprint_covers_holdings_and_actions(
    db_session, employee_user, admin_user,
):
    employee, _ = employee_user
    admin, _ = admin_user
    await _mk_seat(
        db_session, employee.id,
        allocated_at=datetime.now(UTC) - timedelta(days=30),
    )
    db_session.add(AuditLog(
        trace_id=uuid.uuid4(),
        actor_user_id=employee.id,
        actor_type="user",
        action="asset.transfer.confirm",
        status="success",
        resource_type="asset_item",
        resource_id=str(uuid.uuid4()),
        detail={},
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="person_footprint",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    kinds = {p.get("kind") for p in await _payloads(db_session, bundle.id)}
    assert "action" in kinds, "audit rows where they were the actor"
    assert bundle.row_count >= 2


@pytest.mark.asyncio
async def test_footprint_excludes_other_peoples_holdings(
    db_session, employee_user, admin_user,
):
    """The rosters are company-wide; the footprint must narrow to one
    person or it leaks everyone's access into a single-subject bundle."""
    employee, _ = employee_user
    admin, _ = admin_user
    await _mk_seat(
        db_session, employee.id,
        allocated_at=datetime.now(UTC) - timedelta(days=10),
    )
    await _mk_seat(
        db_session, admin.id,
        allocated_at=datetime.now(UTC) - timedelta(days=10),
    )
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="person_footprint",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    rows, _ = await list_bundle_rows(db_session, bundle.id, page_size=200)
    assert rows
    assert all(r.subject_id == str(employee.id) for r in rows)


@pytest.mark.asyncio
async def test_footprint_of_an_unknown_person_declares_a_gap(
    db_session, admin_user,
):
    admin, _ = admin_user
    bundle = await run_evidence_query(
        db_session, query_key="person_footprint",
        params={"person_id": str(uuid.uuid4())},
        actor_user_id=admin.id,
    )
    assert bundle.row_count == 0
    assert bundle.data_gaps[0]["code"] == "person_not_found"


async def _mk_api_key(db, person_id, *, valid_from, valid_until=None):
    """An API-key assignment with its provider — the FK is real."""
    from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider

    provider = ApiKeyProvider(name=f"Provider {uuid.uuid4().hex[:8]}")
    db.add(provider)
    await db.flush()
    item = ApiKeyAssignment(
        provider_id=provider.id,
        assigned_to=person_id,
        request_number=f"AK-{uuid.uuid4().hex[:8]}",
        project_name="Evidence Test",
        reason="revocation-proof coverage",
        monthly_budget_usd=10,
        valid_from=valid_from,
        valid_until=valid_until,
    )
    db.add(item)
    await db.flush()
    return item


# ── revocation proof ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_revocation_proof_reports_what_was_not_revoked(
    db_session, employee_user, admin_user,
):
    """The load-bearing assertion of this whole query.

    Returning only the revoked holdings would make every leaver look
    fully offboarded — a control result manufactured by the query's own
    filter rather than by the platform.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today() - timedelta(days=10)
    await db_session.flush()

    await _mk_seat(
        db_session, employee.id,
        allocated_at=datetime.now(UTC) - timedelta(days=100),
        revoked_at=datetime.now(UTC) - timedelta(days=8),
    )
    # Never revoked — this is the row an audit is looking for.
    await _mk_seat(
        db_session, employee.id,
        allocated_at=datetime.now(UTC) - timedelta(days=100),
    )
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="revocation_proof",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    payloads = await _payloads(db_session, bundle.id)
    seats = [p for p in payloads if p["access"].startswith("software seat")]
    assert sorted(p["still_held"] for p in seats) == [False, True], (
        "a leaver's un-revoked access must appear, not be filtered away"
    )
    # The fixture's own ``employee`` platform role is also still held —
    # correctly reported rather than special-cased away, which is the
    # whole point of not filtering.
    assert any(
        p["still_held"] and p["access"].startswith("platform role")
        for p in payloads
    )


@pytest.mark.asyncio
async def test_revocation_proof_measures_elapsed_days(
    db_session, employee_user, admin_user,
):
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today() - timedelta(days=10)
    await db_session.flush()
    await _mk_seat(
        db_session, employee.id,
        allocated_at=datetime.now(UTC) - timedelta(days=100),
        revoked_at=datetime.combine(
            date.today() - timedelta(days=7), datetime.min.time(), tzinfo=UTC,
        ),
    )
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="revocation_proof",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    seat = next(
        p for p in await _payloads(db_session, bundle.id)
        if p["access"].startswith("software seat")
    )
    assert seat["days_from_termination_to_revocation"] == pytest.approx(
        3.0, abs=0.01,
    )


@pytest.mark.asyncio
async def test_revocation_proof_negative_days_means_revoked_early(
    db_session, employee_user, admin_user,
):
    """Planned offboarding revokes BEFORE the termination date.

    Negative elapsed days is the normal case there, matching the
    convention the platform's other leaver views already use — reading
    it as an anomaly would flood an auditor with false positives.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today()
    await db_session.flush()
    await _mk_seat(
        db_session, employee.id,
        allocated_at=datetime.now(UTC) - timedelta(days=100),
        revoked_at=datetime.combine(
            date.today() - timedelta(days=2), datetime.min.time(), tzinfo=UTC,
        ),
    )
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="revocation_proof",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    seat = next(
        p for p in await _payloads(db_session, bundle.id)
        if p["access"].startswith("software seat")
    )
    assert seat["days_from_termination_to_revocation"] < 0
    assert seat["open_at_termination"] is False


@pytest.mark.asyncio
async def test_revocation_proof_without_termination_date_says_so(
    db_session, employee_user, admin_user,
):
    """No baseline means no elapsed time — and the reader must be told,
    or they will read the output as a clean control result."""
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = None
    await db_session.flush()
    await _mk_seat(
        db_session, employee.id,
        allocated_at=datetime.now(UTC) - timedelta(days=5),
    )
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="revocation_proof",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    assert any(
        g["code"] == "no_termination_date" for g in bundle.data_gaps
    )


@pytest.mark.asyncio
async def test_old_holdings_are_reported_and_flagged_not_filtered_away(
    db_session, employee_user, admin_user,
):
    """No invented cutoff decides what counts as "offboarding-related".

    An earlier version dropped anything revoked before the termination
    date as out of scope, which hid the well-run cases outright: planned
    offboarding revokes a few days EARLY, so a tidy departure produced an
    empty proof. Access given up years ago is reported too — flagged
    ``open_at_termination: False`` so the reader draws the line instead
    of the query drawing it for them, invisibly.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today() - timedelta(days=5)
    await db_session.flush()
    await _mk_seat(
        db_session, employee.id,
        allocated_at=datetime(2024, 1, 1, tzinfo=UTC),
        revoked_at=datetime(2024, 6, 1, tzinfo=UTC),
    )
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="revocation_proof",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    seat = next(
        p for p in await _payloads(db_session, bundle.id)
        if p["access"].startswith("software seat")
    )
    assert seat["open_at_termination"] is False
    assert seat["days_from_termination_to_revocation"] < 0


# ── object lifecycle ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_object_lifecycle_unions_owner_history(
    db_session, employee_user, admin_user,
):
    """Owner changes for network objects never reach audit_logs.

    An audit-only lifecycle query reports "no changes" for a domain that
    changed hands three times — which reads as a clean object rather
    than as a blind spot.

    The two tables also disagree about how to spell the object type:
    ``audit_logs`` writes ``cmdb_domain`` while owner history writes
    ``domain``. So this seeds each row with the value its own producer
    really writes, and asks the question with the audit-log spelling —
    which is the only way the union's vocabulary mapping is actually
    exercised.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    resource_id = str(uuid.uuid4())

    db_session.add(AuditLog(
        trace_id=uuid.uuid4(),
        actor_user_id=admin.id,
        actor_type="user",
        action="cmdb.domain.update",
        status="success",
        resource_type="cmdb_domain",
        resource_id=resource_id,
        detail={"field": "registrar"},
    ))
    db_session.add(CmdbNetworkOwnerHistory(
        # The PRODUCTION value. ``owner_history.RESOURCE_TYPE_DOMAIN``
        # is the canonical short form; only ``audit_logs`` carries the
        # ``cmdb_`` prefix. Seeding this row with the prefixed spelling
        # made the union look like it worked while, against real data,
        # the owner-history half matched nothing at all.
        resource_type="domain",
        resource_id=resource_id,
        previous_owner_person_id=str(employee.id),
        new_owner_person_id=str(admin.id),
        actor_person_id=str(admin.id),
        change_reason="offboarding_transfer",
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="object_lifecycle",
        params={"resource_type": "cmdb_domain", "resource_id": resource_id},
        actor_user_id=admin.id,
    )
    actions = {p["action"] for p in await _payloads(db_session, bundle.id)}
    assert "cmdb.domain.update" in actions
    assert "cmdb.owner_history.offboarding_transfer" in actions
    assert any(
        g["code"] == "owner_changes_live_outside_audit_logs"
        for g in bundle.data_gaps
    )


@pytest.mark.asyncio
async def test_object_lifecycle_rows_cite_the_audit_entry(
    db_session, admin_user,
):
    admin, _ = admin_user
    resource_id = str(uuid.uuid4())
    trace = uuid.uuid4()
    db_session.add(AuditLog(
        trace_id=trace,
        actor_user_id=admin.id,
        actor_type="user",
        action="asset.item.update",
        status="success",
        resource_type="asset_item",
        resource_id=resource_id,
        detail={},
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="object_lifecycle",
        params={"resource_type": "asset_item", "resource_id": resource_id},
        actor_user_id=admin.id,
    )
    rows, _ = await list_bundle_rows(db_session, bundle.id)
    assert rows[0].audit_log_id is not None
    assert rows[0].source_trace_id == trace


# ── access review proof ───────────────────────────────────────────────


async def _mk_campaign(db: AsyncSession, admin_id) -> AccessCertification:
    cert = AccessCertification(
        label=f"Q3 review {uuid.uuid4().hex[:4]}",
        campaign_type="self_attestation",
        status="in_progress",
        started_at=datetime.now(UTC) - timedelta(days=20),
        due_at=datetime.now(UTC) + timedelta(days=10),
        created_by_person_id=admin_id,
    )
    db.add(cert)
    await db.flush()
    return cert


@pytest.mark.asyncio
async def test_review_proof_includes_undecided_items(
    db_session, employee_user, admin_user,
):
    """A campaign reporting only completed items always looks complete."""
    employee, _ = employee_user
    admin, _ = admin_user
    cert = await _mk_campaign(db_session, admin.id)
    now = datetime.now(UTC)

    db_session.add(AccessCertificationItem(
        certification_id=cert.id,
        person_id=employee.id,
        holder_display_name="Reviewed holder",
        access_type="software_seat",
        access_label="Figma Pro",
        assigned_reviewer_person_id=employee.id,
        decision="keep",
        decided_by_person_id=employee.id,
        decided_at=now - timedelta(days=2),
        snapshot_at=now - timedelta(days=20),
    ))
    db_session.add(AccessCertificationItem(
        certification_id=cert.id,
        person_id=employee.id,
        holder_display_name="Undecided holder",
        access_type="cloud_grant",
        access_label="AdministratorAccess",
        assigned_reviewer_person_id=employee.id,
        decision="pending",
        snapshot_at=now - timedelta(days=20),
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="access_review_proof",
        params={"certification_id": str(cert.id)},
        actor_user_id=admin.id,
    )
    decisions = [p["decision"] for p in await _payloads(db_session, bundle.id)]
    assert sorted(decisions) == ["keep", "pending"], (
        "the overdue tail is the finding, not noise to be filtered out"
    )


@pytest.mark.asyncio
async def test_review_proof_carries_the_revocation_trace(
    db_session, employee_user, admin_user,
):
    """A revoke decision without a completed revocation is an intention,
    not a control outcome — the evidence must show which it is."""
    employee, _ = employee_user
    admin, _ = admin_user
    cert = await _mk_campaign(db_session, admin.id)
    now = datetime.now(UTC)
    trace = uuid.uuid4()

    db_session.add(AccessCertificationItem(
        certification_id=cert.id,
        person_id=employee.id,
        holder_display_name="Revoked holder",
        access_type="cloud_grant",
        access_label="AdministratorAccess",
        assigned_reviewer_person_id=employee.id,
        decision="revoke",
        decided_by_person_id=admin.id,
        decided_at=now - timedelta(days=1),
        revocation_status="completed",
        revocation_completed_at=now,
        revocation_trace_id=trace,
        snapshot_at=now - timedelta(days=20),
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="access_review_proof",
        params={"certification_id": str(cert.id)},
        actor_user_id=admin.id,
    )
    rows, _ = await list_bundle_rows(db_session, bundle.id)
    assert rows[0].payload["revocation_status"] == "completed"
    assert rows[0].payload["revocation_trace_id"] == str(trace)
    assert rows[0].source_trace_id == trace


@pytest.mark.asyncio
async def test_review_proof_reports_the_campaign_deadline(
    db_session, employee_user, admin_user,
):
    employee, _ = employee_user
    admin, _ = admin_user
    cert = await _mk_campaign(db_session, admin.id)
    db_session.add(AccessCertificationItem(
        certification_id=cert.id,
        person_id=employee.id,
        holder_display_name="Holder",
        access_type="sso_grant",
        access_label="Grafana",
        assigned_reviewer_person_id=employee.id,
        decision="pending",
        snapshot_at=datetime.now(UTC) - timedelta(days=20),
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="access_review_proof",
        params={"certification_id": str(cert.id)},
        actor_user_id=admin.id,
    )
    payload = (await _payloads(db_session, bundle.id))[0]
    assert payload["due_at"]
    assert payload["campaign"] == cert.label


@pytest.mark.asyncio
async def test_revocation_proof_covers_every_family_the_footprint_does(
    db_session, employee_user, admin_user,
):
    """A proof that omits an access family answers "all clear" for it.

    ``person_footprint`` reads six rosters; the proof read four. So a
    leaver still holding an SSO application or an API key was reported
    as fully offboarded by the proof while their own footprint listed
    the holding — two queries over the same person disagreeing about
    what they had, with the more reassuring one being the one an auditor
    opens first.
    """
    from app.domains.sso.models import SsoGrant

    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today() - timedelta(days=10)
    await db_session.flush()

    db_session.add(SsoGrant(
        person_id=str(employee.id),
        sso_provider="okta",
        system_name="Grafana",
        system_owner_label="Platform Team",
        status="active",
        granted_at=datetime.now(UTC) - timedelta(days=90),
    ))
    await _mk_api_key(
        db_session, employee.id, valid_from=date.today() - timedelta(days=90),
    )
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="revocation_proof",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    payloads = await _payloads(db_session, bundle.id)
    families = {p["access"].split()[0] for p in payloads}
    assert "SSO" in families, "an un-revoked SSO grant must appear"
    assert "API" in families, "an un-revoked API key must appear"
    assert all(
        p["still_held"]
        for p in payloads
        if p["access"].startswith(("SSO application", "API key"))
    )


@pytest.mark.asyncio
async def test_an_api_key_that_lapsed_is_closed_by_its_expiry(
    db_session, employee_user, admin_user,
):
    """An expired key is neither open exposure nor a clean revocation.

    Reporting it as still held overstates the leaver's access; reporting
    it as revoked claims a deliberate act nobody performed. The expiry
    is the close, and the flag says which of the two happened.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today() - timedelta(days=10)
    await db_session.flush()

    await _mk_api_key(
        db_session, employee.id,
        valid_from=date.today() - timedelta(days=90),
        valid_until=date.today() - timedelta(days=30),
    )
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="revocation_proof",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    payloads = await _payloads(db_session, bundle.id)
    key = next(p for p in payloads if p["access"].startswith("API key"))
    assert key["still_held"] is False
    assert key["closed_by_expiry"] is True


@pytest.mark.asyncio
async def test_a_key_valid_until_a_future_date_is_still_held(
    db_session, employee_user, admin_user,
):
    """The branch neither expiry test covered.

    Keys are issued with a validity window and stay active inside it —
    that is what the pre-expiry reminder cron exists for. Treating any
    ``valid_until`` as a close reported a LIVE key as lapsed in the one
    artefact whose job is to surface access that was never revoked, and
    put this query in direct contradiction with ``api_key_roster`` over
    the same row.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today() - timedelta(days=10)
    await db_session.flush()

    await _mk_api_key(
        db_session, employee.id,
        valid_from=date.today() - timedelta(days=90),
        valid_until=date.today() + timedelta(days=180),
    )
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="revocation_proof",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    payloads = await _payloads(db_session, bundle.id)
    key = next(p for p in payloads if p["access"].startswith("API key"))
    assert key["still_held"] is True
    assert key["closed_by_expiry"] is False


@pytest.mark.asyncio
async def test_the_proof_and_the_footprint_agree_on_a_resurrected_grant(
    db_session, employee_user, admin_user,
):
    """The cloud grant row is REUSED on re-request.

    ``UNIQUE(scope, person, role)`` means a re-request overwrites
    ``granted_at`` and clears ``revoked_at``; prior cycles survive only
    in ``detail._resurrection_history``. Reading the live columns alone
    reports one holding where there were two — and makes the proof
    disagree with the footprint, which composes ``cloud_grant_roster``
    and does expand the history.
    """
    from app.domains.cmdb.models import CmdbCloudAccessGrant, CmdbCloudScope
    from app.integrations.models import Integration

    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today() - timedelta(days=10)
    await db_session.flush()

    integration = Integration(
        id=str(uuid.uuid4()), kind="cloud", provider="aws",
        name="proof test", slug=f"pf-{uuid.uuid4().hex[:8]}", enabled=True,
    )
    db_session.add(integration)
    await db_session.flush()
    scope = CmdbCloudScope(
        id=str(uuid.uuid4()), integration_id=str(integration.id),
        scope_type="aws_account", external_id=uuid.uuid4().hex[:12],
        display_name="Proof scope", status="active",
    )
    db_session.add(scope)
    await db_session.flush()

    db_session.add(CmdbCloudAccessGrant(
        scope_id=scope.id,
        person_id=str(employee.id),
        role_key="admin",
        status="active",
        granted_at=datetime.now(UTC) - timedelta(days=20),
        created_at=datetime.now(UTC) - timedelta(days=400),
        detail={"_resurrection_history": [{
            "at": (datetime.now(UTC) - timedelta(days=200)).isoformat(),
            "from_status": "revoked",
            "from_revoked_at": (
                datetime.now(UTC) - timedelta(days=210)
            ).isoformat(),
        }]},
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="revocation_proof",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    payloads = await _payloads(db_session, bundle.id)
    cloud = [p for p in payloads if p["access"].startswith("cloud ")]
    assert len(cloud) == 2, "both holdings, not one"
    assert {p["cycle"] for p in cloud} == {"historical", "current"}
    # The current cycle is un-revoked — the row an audit is looking for.
    assert any(p["still_held"] for p in cloud)


@pytest.mark.asyncio
async def test_a_footprint_mixing_db_and_computed_timestamps_does_not_crash(
    db_session, employee_user, admin_user,
):
    """The most ordinary subject there is, and it 500'd.

    MySQL/TiDB return NAIVE datetimes for DATETIME columns, while
    anything this code constructs is aware — ``datetime.combine(...,
    tzinfo=UTC)`` for an API key's validity dates, ``_parse_iso`` for a
    cloud grant's resurrection history. Composing rosters therefore
    sorted both kinds together and Python raised ``TypeError: can't
    compare offset-naive and offset-aware datetimes``.

    Every existing footprint test happened to use holdings from one
    side of that line only, so the whole suite stayed green while the
    flagship query crashed for any leaver holding an API key plus
    anything else.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    await _mk_seat(
        db_session, employee.id,
        allocated_at=datetime.now(UTC) - timedelta(days=30),
    )
    await _mk_api_key(
        db_session, employee.id,
        valid_from=date.today() - timedelta(days=60),
    )
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="person_footprint",
        params={"person_id": str(employee.id)},
        actor_user_id=admin.id,
    )
    assert bundle.row_count >= 2
    # And the frozen evidence carries an explicit offset rather than a
    # bare local-looking timestamp.
    for payload in await _payloads(db_session, bundle.id):
        assert payload is not None


@pytest.mark.asyncio
async def test_proof_and_roster_agree_on_a_live_key_and_a_disabled_grant(
    db_session, employee_user, admin_user,
):
    """Two queries over one row must not reach opposite conclusions.

    Both disagreements were introduced by fixing one side only: the
    API-key expiry guard went into the proof but not the roster, and the
    cloud close-time reconstruction went into the roster but not the
    proof. Whichever is more reassuring is the one an auditor reads
    first, so a split verdict always favours the wrong answer.
    """
    from app.domains.cmdb.models import CmdbCloudAccessGrant, CmdbCloudScope
    from app.integrations.models import Integration

    employee, _ = employee_user
    admin, _ = admin_user
    employee.termination_date = date.today() - timedelta(days=3)
    await db_session.flush()

    # A key still inside its validity window.
    await _mk_api_key(
        db_session, employee.id,
        valid_from=date.today() - timedelta(days=90),
        valid_until=date.today() + timedelta(days=180),
    )
    # A grant closed by status flip, with no revoke stamp.
    integration = Integration(
        id=str(uuid.uuid4()), kind="cloud", provider="aws",
        name="parity test", slug=f"pt-{uuid.uuid4().hex[:8]}", enabled=True,
    )
    db_session.add(integration)
    await db_session.flush()
    scope = CmdbCloudScope(
        id=str(uuid.uuid4()), integration_id=str(integration.id),
        scope_type="aws_account", external_id=uuid.uuid4().hex[:12],
        display_name="Parity scope", status="active",
    )
    db_session.add(scope)
    await db_session.flush()
    disabled_at = datetime.now(UTC) - timedelta(days=25)
    db_session.add(CmdbCloudAccessGrant(
        scope_id=scope.id, person_id=str(employee.id), role_key="admin",
        status="disabled",
        granted_at=datetime.now(UTC) - timedelta(days=300),
        detail={"disable_result": {"at": disabled_at.isoformat()}},
    ))
    await db_session.flush()

    proof = await _payloads(db_session, (await run_evidence_query(
        db_session, query_key="revocation_proof",
        params={"person_id": str(employee.id)}, actor_user_id=admin.id,
    )).id)
    footprint = await _payloads(db_session, (await run_evidence_query(
        db_session, query_key="person_footprint",
        params={"person_id": str(employee.id)}, actor_user_id=admin.id,
    )).id)

    key_proof = next(p for p in proof if p["access"].startswith("API key"))
    key_roster = next(
        p for p in footprint if p.get("source_table") == "api_key_assignments"
        or "request_number" in p
    )
    assert key_proof["still_held"] is True
    assert key_roster["still_held"] is True

    cloud_proof = next(
        p for p in proof
        if p["access"].startswith("cloud ") and p.get("cycle") == "current"
    )
    cloud_roster = next(
        p for p in footprint if p.get("cycle") == "current" and "role_key" in p
    )
    assert cloud_proof["still_held"] is False, (
        "a grant disabled 25 days ago is not held on the leaver's last day"
    )
    assert cloud_roster["still_held"] is False
    assert cloud_proof["close_time_inferred"] is True