tegwick
b9349782f4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ec5-7e2b-7743-ac08-719e1b0f42e2
141 lines
5.2 KiB
Python
141 lines
5.2 KiB
Python
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
|