import sqlite3 from src.security_event_store import SecurityEventStore def test_security_events_are_private_bounded_and_reverse_chronological(tmp_path): now = [1_000] store = SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: now[0]) store.record( "sign_in", method="token", device_label=" Timmy Phone " + "x" * 80, target="dashboard", ) now[0] += 1 store.record("device_revoked", device_label="Old phone", target="device") page = store.list(limit=1) assert [(event.kind, event.device_label, event.target) for event in page.events] == [ ("device_revoked", "Old phone", "device") ] assert page.next_cursor is not None older = store.list(limit=10, cursor=page.next_cursor) assert older.events[0].kind == "sign_in" assert older.events[0].method == "token" assert older.events[0].device_label == ("Timmy Phone " + "x" * 52) columns = { row[1] for row in sqlite3.connect(store.path).execute("PRAGMA table_info(security_events)") } assert columns == {"id", "kind", "method", "device_label", "target", "created_at"} def test_security_event_retention_prunes_age_and_count(tmp_path): now = [0] store = SecurityEventStore( tmp_path / "security.sqlite3", clock=lambda: now[0], max_events=3, retention_seconds=10, ) for index in range(4): now[0] = index store.record("sign_in", method="token", device_label=f"Device {index}") assert [event.device_label for event in store.list(limit=10).events] == [ "Device 3", "Device 2", "Device 1" ] now[0] = 20 store.record("sign_out", device_label="Current") assert [event.kind for event in store.list(limit=10).events] == ["sign_out"]