"""Evidence query catalogue + frozen bundles.

The claims this feature makes to an external auditor are: the answer is
reproducible, every line is checkable, and incompleteness is declared
rather than hidden. These tests exist to keep those three claims true —
they are the difference between an evidence tool and a screenshot.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthRole, AuthUserRole
from app.core.access_cycle_listener import register_listener
from app.core.exceptions import AppException
from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.auditor.evidence_query import catalog
from app.domains.auditor.evidence_query.contracts import (
    EvidenceRow,
    canonical_params,
    content_hash,
)
from app.domains.auditor.evidence_query.service import (
    _run_time_upper_bound,
    list_bundle_rows,
    run_evidence_query,
)
from app.domains.auditor.models import AuditorEvidenceBundleRow

register_listener()


async def _mk_software_with_plan(
    db: AsyncSession,
) -> tuple[AssetSoftwareCatalog, SoftwareLicensePlan]:
    software = AssetSoftwareCatalog(name=f"EvidenceSW {uuid.uuid4().hex[:6]}")
    db.add(software)
    await db.flush()
    plan = SoftwareLicensePlan(
        software_id=software.id,
        name="Pro",
        price=100,
        billing_cycle="monthly",
    )
    db.add(plan)
    await db.flush()
    return software, plan


# ── catalogue contract ────────────────────────────────────────────────


def test_catalog_validates_at_import_time():
    """A malformed registration must fail the process, not the auditor.

    ``validate_catalog()`` runs on import; calling it directly here
    proves the check is real rather than a no-op that happens to pass.
    """
    catalog.validate_catalog()
    assert catalog.EVIDENCE_QUERIES


def test_every_query_declares_all_three_languages():
    for definition in catalog.EVIDENCE_QUERIES:
        public = definition.public()
        assert set(public["title"]) == {"zh-CN", "en-US", "ja-JP"}
        assert all(public["title"].values()), definition.key
        for param in public["params"]:
            assert all(param["label"].values()), (definition.key, param["name"])


def test_catalog_rejects_duplicate_keys(monkeypatch):
    """Mutation check on the validator itself."""
    duplicated = catalog.EVIDENCE_QUERIES + (catalog.EVIDENCE_QUERIES[0],)
    monkeypatch.setattr(catalog, "EVIDENCE_QUERIES", duplicated)
    with pytest.raises(RuntimeError, match="duplicate evidence query key"):
        catalog.validate_catalog()


# ── reproducibility ───────────────────────────────────────────────────


def test_canonical_params_normalise_dates_and_key_order():
    """Same question, two spellings, one canonical form.

    Without this the same question asked two ways would produce two
    bundles that cannot be compared, and the reproducibility claim would
    be hollow.
    """
    from datetime import date

    a = canonical_params({"period_end": date(2026, 5, 31), "period_start": "2026-05-01"})
    b = canonical_params({"period_start": "2026-05-01", "period_end": "2026-05-31"})
    assert a == b
    assert list(a) == ["period_end", "period_start"]


def test_content_hash_ignores_bundle_identity_but_tracks_facts():
    base = EvidenceRow(
        subject_type="person",
        subject_id="p1",
        source_table="t",
        source_pk="1",
        payload={"a": 1},
    )
    same = EvidenceRow(
        subject_type="person",
        subject_id="p1",
        source_table="t",
        source_pk="1",
        payload={"a": 1},
    )
    changed = EvidenceRow(
        subject_type="person",
        subject_id="p1",
        source_table="t",
        source_pk="1",
        payload={"a": 2},
    )
    assert content_hash([base]) == content_hash([same])
    assert content_hash([base]) != content_hash([changed])
    # Order is part of the hash — the rows are stored in emission order
    # and a reader reproduces the hash by walking ``seq``.
    assert content_hash([base, changed]) != content_hash([changed, base])


@pytest.mark.asyncio
async def test_rerunning_the_same_question_reproduces_the_hash(
    db_session, employee_user, admin_user,
):
    employee, _ = employee_user
    admin, _ = admin_user
    software, plan = await _mk_software_with_plan(db_session)
    db_session.add(SoftwareSeatAllocation(
        plan_id=plan.id,
        user_id=employee.id,
        status="active",
        allocated_at=datetime.now(UTC) - timedelta(days=20),
    ))
    await db_session.flush()

    params = {
        "software_id": str(software.id),
        "period_start": "2026-01-01",
        "period_end": "2026-12-31",
    }
    first = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params=dict(params), actor_user_id=admin.id,
    )
    second = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params=dict(params), actor_user_id=admin.id,
    )

    assert first.id != second.id, "a re-run must freeze a NEW bundle"
    assert first.content_hash == second.content_hash
    assert first.row_count == second.row_count == 1


@pytest.mark.asyncio
async def test_hash_changes_when_the_underlying_facts_move(
    db_session, employee_user, admin_user,
):
    """A differing hash is the signal, not a defect.

    Two runs of the same question at the same version disagreeing means
    the underlying data changed between them — which is itself an audit
    finding.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    software, plan = await _mk_software_with_plan(db_session)
    seat = SoftwareSeatAllocation(
        plan_id=plan.id,
        user_id=employee.id,
        status="active",
        allocated_at=datetime.now(UTC) - timedelta(days=20),
    )
    db_session.add(seat)
    await db_session.flush()

    params = {
        "software_id": str(software.id),
        "period_start": "2026-01-01",
        "period_end": "2026-12-31",
    }
    before = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params=dict(params), actor_user_id=admin.id,
    )

    seat.status = "revoked"
    seat.revoked_at = datetime.now(UTC)
    await db_session.flush()

    after = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params=dict(params), actor_user_id=admin.id,
    )
    assert before.content_hash != after.content_hash


def test_default_upper_bound_is_a_whole_second_not_the_exact_instant():
    """Regression: MySQL rounds sub-second timestamps UP on insert.

    Most lifecycle columns are second-granular ``DATETIME``, and MySQL
    rounds rather than truncates — verified directly: inserting
    ``12:00:01.618`` stores ``12:00:02``. An auto-filled upper bound of
    the exact current instant therefore sorts *before* a grant that was
    just made, and the grant drops out of its own roster. "The row I
    created a second ago is missing" is the fastest possible way to lose
    an auditor's trust in the tool.

    Asserted on the bound itself rather than through a live query: the
    end-to-end symptom only fires when the sub-second margin happens to
    land the wrong way, which reads as flake rather than as the bug it
    is.
    """
    before = datetime.now(UTC)
    bound = _run_time_upper_bound()
    assert bound.microsecond == 0, (
        "an exact-instant bound excludes rows MySQL rounded up on insert"
    )
    assert bound >= before.replace(microsecond=0)
    # At most one second of look-ahead, which cannot include anything
    # that has not already happened.
    assert (bound - before).total_seconds() <= 1


@pytest.mark.asyncio
async def test_a_grant_rounded_up_by_mysql_is_still_inside_the_window(
    db_session, employee_user, admin_user,
):
    """End-to-end companion to the bound test above.

    ``.9`` of a second is forced so the stored value is guaranteed to
    round into the NEXT second — the exact case an exact-instant bound
    would miss.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    software, plan = await _mk_software_with_plan(db_session)
    db_session.add(SoftwareSeatAllocation(
        plan_id=plan.id,
        user_id=employee.id,
        status="active",
        allocated_at=datetime.now(UTC).replace(microsecond=900000),
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params={"software_id": str(software.id)},
        actor_user_id=admin.id,
    )
    assert bundle.row_count == 1


# ── provenance ────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_every_persisted_row_cites_a_record(
    db_session, employee_user, admin_user,
):
    employee, _ = employee_user
    admin, _ = admin_user
    software, plan = await _mk_software_with_plan(db_session)
    seat = SoftwareSeatAllocation(
        plan_id=plan.id,
        user_id=employee.id,
        status="active",
        allocated_at=datetime.now(UTC) - timedelta(days=5),
    )
    db_session.add(seat)
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params={"software_id": str(software.id)},
        actor_user_id=admin.id,
    )
    rows, total = await list_bundle_rows(db_session, bundle.id)
    assert total == 1
    assert rows[0].source_table == "software_seat_allocations"
    assert rows[0].source_pk == str(seat.id)
    assert rows[0].subject_label  # resolvable back to a human


@pytest.mark.asyncio
async def test_a_row_without_provenance_is_refused(
    db_session, admin_user, monkeypatch,
):
    """An unciteable line is worse than a missing one — a reader cannot
    tell it apart from a checkable one."""
    admin, _ = admin_user
    from dataclasses import replace

    from app.domains.auditor.evidence_query.contracts import EvidenceResult

    async def _bad_handler(_db, _params):
        return EvidenceResult(rows=[
            EvidenceRow(subject_type="person", payload={"claim": "trust me"}),
        ])

    # Definitions are frozen on purpose — swap in a replaced copy rather
    # than mutating the registered one.
    bad = replace(catalog.get_query("software_seat_roster"), handler=_bad_handler)
    monkeypatch.setitem(catalog._BY_KEY, "software_seat_roster", bad)

    with pytest.raises(AppException) as exc:
        await run_evidence_query(
            db_session, query_key="software_seat_roster",
            params={"software_id": str(uuid.uuid4())},
            actor_user_id=admin.id,
        )
    assert exc.value.code == "EVIDENCE_ROW_WITHOUT_PROVENANCE"


# ── parameter handling ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_unknown_parameter_is_refused_not_ignored(db_session, admin_user):
    """A silently dropped typo widens the roster with no visible sign."""
    admin, _ = admin_user
    with pytest.raises(AppException) as exc:
        await run_evidence_query(
            db_session, query_key="software_seat_roster",
            params={"software_id": str(uuid.uuid4()), "periodstart": "2026-05-01"},
            actor_user_id=admin.id,
        )
    assert exc.value.code == "EVIDENCE_QUERY_UNKNOWN_PARAM"


@pytest.mark.asyncio
async def test_missing_required_parameter_is_refused(db_session, admin_user):
    """``person_footprint`` genuinely needs a subject.

    (The seat roster's ``software_id`` became optional once the
    person-footprint query started composing it — a period-wide seat
    roster is a legitimate question in its own right.)
    """
    admin, _ = admin_user
    with pytest.raises(AppException) as exc:
        await run_evidence_query(
            db_session, query_key="person_footprint",
            params={}, actor_user_id=admin.id,
        )
    assert exc.value.code == "EVIDENCE_QUERY_MISSING_PARAM"


@pytest.mark.asyncio
async def test_unregistered_query_key_is_404(db_session, admin_user):
    admin, _ = admin_user
    with pytest.raises(AppException) as exc:
        await run_evidence_query(
            db_session, query_key="drop_table_users",
            params={}, actor_user_id=admin.id,
        )
    assert exc.value.code == "EVIDENCE_QUERY_NOT_FOUND"


# ── handler semantics ─────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_period_roster_returns_holdings_that_span_the_window(
    db_session, employee_user, admin_user,
):
    """The population an auditor most wants is the one a naive
    ``granted_at BETWEEN`` filter silently drops."""
    employee, _ = employee_user
    admin, _ = admin_user
    software, plan = await _mk_software_with_plan(db_session)
    # Allocated long before the window, never revoked → held throughout.
    db_session.add(SoftwareSeatAllocation(
        plan_id=plan.id,
        user_id=employee.id,
        status="active",
        allocated_at=datetime(2025, 1, 1, tzinfo=UTC),
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params={
            "software_id": str(software.id),
            "period_start": "2026-05-01",
            "period_end": "2026-05-31",
        },
        actor_user_id=admin.id,
    )
    assert bundle.row_count == 1


@pytest.mark.asyncio
async def test_seat_revoked_before_the_window_is_excluded(
    db_session, employee_user, admin_user,
):
    employee, _ = employee_user
    admin, _ = admin_user
    software, plan = await _mk_software_with_plan(db_session)
    db_session.add(SoftwareSeatAllocation(
        plan_id=plan.id,
        user_id=employee.id,
        status="revoked",
        allocated_at=datetime(2025, 1, 1, tzinfo=UTC),
        revoked_at=datetime(2025, 6, 1, tzinfo=UTC),
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params={
            "software_id": str(software.id),
            "period_start": "2026-05-01",
            "period_end": "2026-05-31",
        },
        actor_user_id=admin.id,
    )
    assert bundle.row_count == 0


@pytest.mark.asyncio
async def test_privileged_role_roster_reads_the_holding_ledger(
    db_session, employee_user, admin_user,
):
    employee, _ = employee_user
    admin, _ = admin_user
    role_id = (
        await db_session.execute(
            select(AuthRole.id).where(AuthRole.name == "auditor")
        )
    ).scalar_one()

    link = AuthUserRole(user_id=employee.id, role_id=role_id)
    db_session.add(link)
    await db_session.flush()

    held = await run_evidence_query(
        db_session, query_key="privileged_role_roster",
        params={"role_name": "auditor"},
        actor_user_id=admin.id,
    )
    subjects = {
        r.subject_id
        for r in (await list_bundle_rows(db_session, held.id))[0]
    }
    assert str(employee.id) in subjects

    # Revoke, then ask again "as of now" — the holding must drop out
    # even though auth_user_roles has no revoked_at at all.
    await db_session.delete(link)
    await db_session.flush()

    after = await run_evidence_query(
        db_session, query_key="privileged_role_roster",
        params={"role_name": "auditor"},
        actor_user_id=admin.id,
    )
    subjects_after = {
        r.subject_id
        for r in (await list_bundle_rows(db_session, after.id))[0]
    }
    assert str(employee.id) not in subjects_after


@pytest.mark.asyncio
async def test_range_predating_the_ledger_declares_a_data_gap(
    db_session, admin_user,
):
    """Incompleteness is declared, never silent.

    An auditor told "this range predates the ledger" can still work; one
    handed a quietly short list cannot.
    """
    admin, _ = admin_user
    bundle = await run_evidence_query(
        db_session, query_key="privileged_role_roster",
        params={"role_name": "auditor", "at": "2020-01-01"},
        actor_user_id=admin.id,
    )
    assert bundle.data_gaps
    assert bundle.data_gaps[0]["code"] == "role_ledger_starts_at_migration_484"


# ── auditability of the tool itself ───────────────────────────────────


@pytest.mark.asyncio
async def test_running_a_query_is_itself_audited(
    db_session, employee_user, admin_user,
):
    """An access-forensics tool that leaves no trace of who searched for
    whom is a governance hole in its own right."""
    from app.audit.models import AuditLog

    employee, _ = employee_user
    admin, _ = admin_user
    software, plan = await _mk_software_with_plan(db_session)
    db_session.add(SoftwareSeatAllocation(
        plan_id=plan.id, user_id=employee.id, status="active",
        allocated_at=datetime.now(UTC),
    ))
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params={"software_id": str(software.id)},
        actor_user_id=admin.id,
    )

    log = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "auditor.evidence_query.run",
                AuditLog.resource_id == str(bundle.id),
            )
        )
    ).scalar_one()
    assert log.actor_user_id == admin.id
    assert log.detail["query_key"] == "software_seat_roster"
    assert log.detail["content_hash"] == bundle.content_hash
    # The parameters are recorded too: "who looked up whom" is the part
    # that matters when the forensics tool itself is under review.
    assert log.detail["params"]["software_id"] == str(software.id)


@pytest.mark.asyncio
async def test_rows_are_frozen_not_recomputed(
    db_session, employee_user, admin_user,
):
    """Delete the underlying record; the bundle must still read the same.

    This is the property that makes a bundle usable months later.
    """
    employee, _ = employee_user
    admin, _ = admin_user
    software, plan = await _mk_software_with_plan(db_session)
    seat = SoftwareSeatAllocation(
        plan_id=plan.id, user_id=employee.id, status="active",
        allocated_at=datetime.now(UTC),
    )
    db_session.add(seat)
    await db_session.flush()

    bundle = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params={"software_id": str(software.id)},
        actor_user_id=admin.id,
    )
    await db_session.delete(seat)
    await db_session.flush()

    rows = (
        await db_session.execute(
            select(AuditorEvidenceBundleRow)
            .where(AuditorEvidenceBundleRow.bundle_id == bundle.id)
        )
    ).scalars().all()
    assert len(rows) == 1
    assert rows[0].payload["status"] == "active"


@pytest.mark.asyncio
async def test_hash_is_reproducible_from_the_persisted_rows(
    db_session, employee_user, admin_user,
):
    """The IPE claim, verified end to end rather than asserted.

    A reviewer who rebuilds ``EvidenceRow`` objects from
    ``auditor_evidence_bundle_rows`` and re-computes ``content_hash``
    must get the stored digest back. That only holds if every field the
    digest covers survives storage at the same precision — which it did
    NOT at first: ``occurred_at`` was hashed with microseconds (the
    holding ledgers use ``DATETIME(6)``) but persisted into a
    whole-second column, and MySQL rounds sub-second inserts UP. The
    bundle would still have looked internally fine; only re-deriving the
    hash exposes the divergence.

    Uses ``privileged_role_roster`` deliberately: its ``occurred_at``
    comes straight from a microsecond-precision ledger column.
    """
    employee, _ = employee_user
    admin, _ = admin_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()

    bundle = await run_evidence_query(
        db_session, query_key="privileged_role_roster",
        params={"role_name": "auditor"},
        actor_user_id=admin.id,
    )
    stored, _total = await list_bundle_rows(db_session, bundle.id, page_size=500)
    assert stored

    rebuilt = [
        EvidenceRow(
            subject_type=row.subject_type,
            subject_id=row.subject_id,
            subject_label=row.subject_label,
            source_table=row.source_table,
            source_pk=row.source_pk,
            audit_log_id=row.audit_log_id,
            occurred_at=row.occurred_at,
            payload=row.payload,
        )
        for row in stored
    ]
    assert content_hash(rebuilt) == bundle.content_hash