108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
"""Tests for AuditLog."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from atm.audit import AuditLog
|
|
|
|
|
|
def _dt(s: str) -> datetime:
|
|
return datetime.fromisoformat(s)
|
|
|
|
|
|
def test_writes_jsonl(tmp_path: Path) -> None:
|
|
clock_dt = _dt("2026-04-15T10:00:00")
|
|
log = AuditLog(tmp_path / "logs", clock=lambda: clock_dt)
|
|
|
|
events = [
|
|
{"msg": "first", "ts": "2026-04-15T10:00:00"},
|
|
{"msg": "second", "ts": "2026-04-15T10:01:00"},
|
|
{"msg": "third", "ts": "2026-04-15T10:02:00"},
|
|
]
|
|
for e in events:
|
|
log.log(e)
|
|
log.close()
|
|
|
|
lines = (tmp_path / "logs" / "2026-04-15.jsonl").read_text().splitlines()
|
|
assert len(lines) == 3
|
|
for original, line in zip(events, lines):
|
|
parsed = json.loads(line)
|
|
assert parsed["msg"] == original["msg"]
|
|
assert parsed["ts"] == original["ts"]
|
|
|
|
|
|
def test_daily_rotation(tmp_path: Path) -> None:
|
|
base = tmp_path / "logs"
|
|
times = [
|
|
_dt("2026-04-15T23:59:59"), # just before midnight
|
|
_dt("2026-04-16T00:00:01"), # just after midnight
|
|
]
|
|
idx = 0
|
|
|
|
def clock() -> datetime:
|
|
return times[min(idx, len(times) - 1)]
|
|
|
|
log = AuditLog(base, clock=clock)
|
|
|
|
log.log({"msg": "before"})
|
|
idx = 1
|
|
log.log({"msg": "after"})
|
|
log.close()
|
|
|
|
file_15 = base / "2026-04-15.jsonl"
|
|
file_16 = base / "2026-04-16.jsonl"
|
|
assert file_15.exists(), "File for Apr 15 should exist"
|
|
assert file_16.exists(), "File for Apr 16 should exist"
|
|
|
|
lines_15 = [json.loads(l) for l in file_15.read_text().splitlines()]
|
|
lines_16 = [json.loads(l) for l in file_16.read_text().splitlines()]
|
|
assert lines_15[0]["msg"] == "before"
|
|
assert lines_16[0]["msg"] == "after"
|
|
|
|
|
|
def test_line_buffered(tmp_path: Path) -> None:
|
|
base = tmp_path / "logs"
|
|
clock_dt = _dt("2026-04-15T12:00:00")
|
|
log = AuditLog(base, clock=lambda: clock_dt)
|
|
log.log({"msg": "hello", "ts": "2026-04-15T12:00:00"})
|
|
# Do NOT call close() — file should already have content due to line buffering
|
|
path = base / "2026-04-15.jsonl"
|
|
assert os.stat(path).st_size > 0
|
|
log.close()
|
|
|
|
|
|
def test_adds_ts_when_missing(tmp_path: Path) -> None:
|
|
clock_dt = _dt("2026-04-15T09:30:00")
|
|
log = AuditLog(tmp_path / "logs", clock=lambda: clock_dt)
|
|
log.log({"msg": "hi"})
|
|
log.close()
|
|
|
|
lines = (tmp_path / "logs" / "2026-04-15.jsonl").read_text().splitlines()
|
|
parsed = json.loads(lines[0])
|
|
assert "ts" in parsed
|
|
assert parsed["ts"] == "2026-04-15T09:30:00"
|
|
|
|
|
|
def test_preserves_existing_ts(tmp_path: Path) -> None:
|
|
clock_dt = _dt("2026-04-15T09:30:00")
|
|
log = AuditLog(tmp_path / "logs", clock=lambda: clock_dt)
|
|
log.log({"ts": "2026-01-01T00:00:00", "msg": "hi"})
|
|
log.close()
|
|
|
|
lines = (tmp_path / "logs" / "2026-04-15.jsonl").read_text().splitlines()
|
|
parsed = json.loads(lines[0])
|
|
assert parsed["ts"] == "2026-01-01T00:00:00"
|
|
|
|
|
|
def test_close_idempotent(tmp_path: Path) -> None:
|
|
clock_dt = _dt("2026-04-15T10:00:00")
|
|
log = AuditLog(tmp_path / "logs", clock=lambda: clock_dt)
|
|
log.log({"msg": "x", "ts": "2026-04-15T10:00:00"})
|
|
log.close()
|
|
log.close() # should not raise
|