55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
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"]
|