170 lines
5.0 KiB
Python
170 lines
5.0 KiB
Python
"""Tests for the pure ``operations.audit`` layer (W3 of the web-UI plan).
|
|
|
|
Covers ``append_audit``/``read_audit`` — the append-only JSONL trail the
|
|
future web UI writes to on every mutating action. Zero ``click``/``print``
|
|
in the module under test; only stdlib ``json`` is used.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from ci_orchestrator.operations.audit import AuditEntry, append_audit, read_audit
|
|
|
|
|
|
def _entry(**overrides: object) -> AuditEntry:
|
|
base: dict[str, object] = {
|
|
"ts": "2026-07-23T12:00:00+00:00",
|
|
"user": "simone",
|
|
"action": "vm.remove",
|
|
"target": "CI-build-abc123",
|
|
"result": "ok",
|
|
"detail": "removed via web UI",
|
|
}
|
|
base.update(overrides)
|
|
return AuditEntry(**base) # type: ignore[arg-type]
|
|
|
|
|
|
# ── append_audit / read_audit round-trip ────────────────────────────────────
|
|
|
|
|
|
def test_append_then_read_round_trip(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "audit" / "web-audit.jsonl"
|
|
entry = _entry()
|
|
|
|
append_audit(log_path, entry)
|
|
result = read_audit(log_path)
|
|
|
|
assert result == [entry]
|
|
|
|
|
|
def test_append_creates_parent_dir(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "nested" / "dir" / "web-audit.jsonl"
|
|
|
|
append_audit(log_path, _entry())
|
|
|
|
assert log_path.is_file()
|
|
|
|
|
|
def test_append_writes_one_json_line(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "web-audit.jsonl"
|
|
entry = _entry()
|
|
|
|
append_audit(log_path, entry)
|
|
|
|
lines = log_path.read_text(encoding="utf-8").splitlines()
|
|
assert len(lines) == 1
|
|
parsed = json.loads(lines[0])
|
|
assert parsed == {
|
|
"ts": entry.ts,
|
|
"user": entry.user,
|
|
"action": entry.action,
|
|
"target": entry.target,
|
|
"result": entry.result,
|
|
"detail": entry.detail,
|
|
}
|
|
|
|
|
|
def test_append_never_truncates(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "web-audit.jsonl"
|
|
first = _entry(target="vm-1")
|
|
second = _entry(target="vm-2")
|
|
|
|
append_audit(log_path, first)
|
|
append_audit(log_path, second)
|
|
|
|
result = read_audit(log_path)
|
|
assert result == [first, second]
|
|
lines = log_path.read_text(encoding="utf-8").splitlines()
|
|
assert len(lines) == 2
|
|
|
|
|
|
# ── read_audit ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_read_audit_missing_file_returns_empty_list(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "does-not-exist.jsonl"
|
|
|
|
assert read_audit(log_path) == []
|
|
|
|
|
|
def test_read_audit_returns_last_n_in_file_order(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "web-audit.jsonl"
|
|
entries = [_entry(target=f"vm-{i}") for i in range(5)]
|
|
for entry in entries:
|
|
append_audit(log_path, entry)
|
|
|
|
result = read_audit(log_path, last=3)
|
|
|
|
assert result == entries[-3:]
|
|
|
|
|
|
def test_read_audit_default_last_is_100(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "web-audit.jsonl"
|
|
entries = [_entry(target=f"vm-{i}") for i in range(150)]
|
|
for entry in entries:
|
|
append_audit(log_path, entry)
|
|
|
|
result = read_audit(log_path)
|
|
|
|
assert len(result) == 100
|
|
assert result == entries[-100:]
|
|
|
|
|
|
def test_read_audit_skips_malformed_lines(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "web-audit.jsonl"
|
|
good = _entry(target="vm-good")
|
|
append_audit(log_path, good)
|
|
with log_path.open("a", encoding="utf-8") as fh:
|
|
fh.write("{not valid json torn line\n")
|
|
fh.write("\n") # blank line, also skipped
|
|
|
|
result = read_audit(log_path)
|
|
|
|
assert result == [good]
|
|
|
|
|
|
def test_read_audit_skips_non_object_json_line(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "web-audit.jsonl"
|
|
good = _entry(target="vm-good")
|
|
append_audit(log_path, good)
|
|
with log_path.open("a", encoding="utf-8") as fh:
|
|
fh.write("[1, 2, 3]\n") # valid JSON, but not an object
|
|
fh.write('"just a string"\n')
|
|
|
|
result = read_audit(log_path)
|
|
|
|
assert result == [good]
|
|
|
|
|
|
def test_read_audit_skips_object_missing_required_key(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "web-audit.jsonl"
|
|
good = _entry(target="vm-good")
|
|
append_audit(log_path, good)
|
|
with log_path.open("a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps({"ts": "x", "user": "simone"}) + "\n") # missing keys
|
|
|
|
result = read_audit(log_path)
|
|
|
|
assert result == [good]
|
|
|
|
|
|
def test_read_audit_last_zero_returns_empty_list(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "web-audit.jsonl"
|
|
append_audit(log_path, _entry())
|
|
|
|
assert read_audit(log_path, last=0) == []
|
|
|
|
|
|
def test_read_audit_skips_torn_last_line(tmp_path: Path) -> None:
|
|
log_path = tmp_path / "web-audit.jsonl"
|
|
good = _entry(target="vm-good")
|
|
append_audit(log_path, good)
|
|
with log_path.open("a", encoding="utf-8") as fh:
|
|
fh.write('{"ts": "2026-07-23T12:00:01+00:00", "user": "simone", "acti')
|
|
|
|
result = read_audit(log_path)
|
|
|
|
assert result == [good]
|