import asyncio
import logging
from concurrent.futures import ThreadPoolExecutor
from contextlib import suppress

from qonto_assistant.app import _emit_audit_heartbeats
from qonto_assistant.audit import AUDIT_STREAM_ID, REDACTED, AuditLogger


def test_audit_logger_redacts_secret_fields() -> None:
    events: list[dict[str, object]] = []
    logger = logging.getLogger("qonto_assistant.audit.test")
    logger.handlers.clear()
    audit = AuditLogger(logger=logger, sink=events.append)

    payload = audit.emit(
        {
            "authorization": "Bearer super-secret",
            "api_key": "top-secret",
            "nested": {"token": "child-secret"},
            "capability": "org_summary",
        }
    )

    assert payload["authorization"] == REDACTED
    assert payload["api_key"] == REDACTED
    assert payload["nested"]["token"] == REDACTED
    assert "super-secret" not in str(events[0])
    assert "top-secret" not in str(events[0])


def test_audit_logger_sequences_events_and_counts_source_transitions() -> None:
    events: list[dict[str, object]] = []
    audit = AuditLogger(sink=events.append, instance_id="instance-1")

    allowed = audit.emit({"decision": "allow", "request_id": "allow-1"})
    denied = audit.emit({"decision": "deny", "request_id": "deny-1"})
    snapshot = audit.reconciliation_snapshot()

    assert allowed["event_class"] == "audit.allow"
    assert denied["event_class"] == "audit.deny"
    assert [event["stream_sequence"] for event in events] == [1, 2]
    assert all(event["stream_id"] == AUDIT_STREAM_ID for event in events)
    assert all(event["stream_instance_id"] == "instance-1" for event in events)
    assert snapshot["last_stream_sequence"] == 2
    assert snapshot["source_transition_counts"] == {"audit.allow": 1, "audit.deny": 1}
    assert snapshot["since_last_heartbeat_counts"] == {"audit.allow": 1, "audit.deny": 1}


def test_heartbeat_reconciles_window_and_resets_only_window_counts() -> None:
    timestamps = iter(
        [
            "2026-09-04T00:00:00+00:00",
            "2026-09-04T00:01:00+00:00",
            "2026-09-04T00:02:00+00:00",
            "2026-09-04T00:03:00+00:00",
        ]
    )
    events: list[dict[str, object]] = []
    audit = AuditLogger(
        sink=events.append, instance_id="instance-1", clock=lambda: next(timestamps)
    )

    first = audit.emit_heartbeat(reason="startup")
    audit.emit({"decision": "deny", "request_id": "deny-1"})
    second = audit.emit_heartbeat(reason="periodic")
    snapshot = audit.reconciliation_snapshot()

    assert first["assertion"] == "nothing-to-report"
    assert first["window_transition_counts"] == {"audit.allow": 0, "audit.deny": 0}
    assert second["assertion"] == "transitions-reported"
    assert second["source_transition_counts"] == {"audit.allow": 0, "audit.deny": 1}
    assert second["window_transition_counts"] == {"audit.allow": 0, "audit.deny": 1}
    assert second["stream_sequence"] == 3
    assert snapshot["source_transition_counts"] == {"audit.allow": 0, "audit.deny": 1}
    assert snapshot["since_last_heartbeat_counts"] == {"audit.allow": 0, "audit.deny": 0}


def test_primary_log_is_written_before_optional_sink_failure(caplog) -> None:
    def broken_sink(_: dict[str, object]) -> None:
        raise RuntimeError("secondary sink unavailable")

    audit = AuditLogger(sink=broken_sink, instance_id="instance-1")

    with caplog.at_level(logging.INFO, logger="qonto_assistant.audit"):
        try:
            audit.emit({"decision": "deny", "request_id": "deny-1"})
        except RuntimeError:
            pass
        else:
            raise AssertionError("Expected the secondary sink error")

    assert '"event_class": "audit.deny"' in caplog.text
    assert '"stream_sequence": 1' in caplog.text


async def test_periodic_heartbeat_loop_emits_until_cancelled() -> None:
    events: list[dict[str, object]] = []
    audit = AuditLogger(sink=events.append, instance_id="instance-1")
    ready = asyncio.Event()

    def collect(payload):
        events.append(payload)
        if len(events) >= 2:
            ready.set()

    audit.sink = collect
    task = asyncio.create_task(_emit_audit_heartbeats(audit_logger=audit, interval_seconds=0.01))
    try:
        await asyncio.wait_for(ready.wait(), timeout=2)
    finally:
        task.cancel()
        with suppress(asyncio.CancelledError):
            await task

    assert len(events) >= 2
    assert all(event["event_class"] == "audit.heartbeat" for event in events)
    assert all(event["reason"] == "periodic" for event in events)


def test_concurrent_transitions_and_heartbeats_preserve_publication_order() -> None:
    events = []
    audit = AuditLogger(sink=events.append)

    def publish(index):
        if index % 3 == 0:
            audit.emit_heartbeat()
        else:
            audit.emit({"decision": "deny"})

    with ThreadPoolExecutor(max_workers=8) as executor:
        list(executor.map(publish, range(120)))

    assert [event["stream_sequence"] for event in events] == list(range(1, 121))
    denies = 0
    for event in events:
        if event["event_class"] == "audit.deny":
            denies += 1
        else:
            assert event["source_transition_counts"]["audit.deny"] == denies
    assert audit.reconciliation_snapshot()["source_transition_counts"]["audit.deny"] == 80