Spaces:
Sleeping
Sleeping
File size: 1,319 Bytes
58ce8cf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 | from app.audit import build_row
def test_build_row():
row = build_row(session="abc", text="i hate you",
read=[65,148,81,0,128,93,128],
mood_before=[128]*7, mood_after=[120]*7,
deltas=[-8,0,0,0,0,0,0],
trace={"words": [], "structures": [], "contributors": [],
"unknown_tokens": ["mid"], "suspected_gap": False},
mood_word="stung", ts=1781500000)
assert row["text"] == "i hate you"
assert row["unknown_tokens"] == ["mid"]
assert row["read"][0] == 65 and len(row["mood_after"]) == 7
from app.audit import AuditLog
def test_buffer_accumulates(tmp_path, monkeypatch):
log = AuditLog(repo_id="x/y", local_dir=str(tmp_path), flush_every=3)
flushed = {}
monkeypatch.setattr(log, "_commit", lambda rows: flushed.setdefault("n", len(rows)))
for _ in range(3):
log.append({"ts": 1, "text": "hi"})
assert flushed.get("n") == 3 # flushed at threshold
assert log.pending() == 0
def test_append_never_raises(tmp_path, monkeypatch):
log = AuditLog(repo_id="x/y", local_dir=str(tmp_path), flush_every=1)
monkeypatch.setattr(log, "_commit", lambda rows: (_ for _ in ()).throw(RuntimeError("net")))
log.append({"ts": 1}) # must not raise
|