133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
"""The append-only store behind logg.jsonl, beslutninger.jsonl, varianter.jsonl
|
|
(plan Step 5).
|
|
|
|
Three risks meet in this one module. Risk H4: the CLI and the host MCP server
|
|
can both be writing while the operator watches, so an append that is not a
|
|
single O_APPEND write can interleave two records into one unreadable line.
|
|
Risk H10: the store is append-only, so its schema is irreversible -- `type`
|
|
and `skjema` have to be on the very first line ever written, because there is
|
|
no migration that can add them later. Risk M5r: a torn trailing line from a
|
|
killed process is recoverable and must not take the whole log with it, while a
|
|
malformed line in the middle is corruption and must stop the run loudly.
|
|
|
|
The DST test is the one that looks like trivia and is not. Norwegian local
|
|
time repeats 02:30 on the last Sunday of October; two records written an hour
|
|
apart carry the same wall clock and differ only in the offset. Sorting on the
|
|
string would call them simultaneous.
|
|
|
|
Style note: this file follows tests/test_frontmatter.py.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from jobbsok_lib import jsonl
|
|
|
|
|
|
def entry(**overrides):
|
|
record = {
|
|
"type": "beslutning",
|
|
"id": "b1",
|
|
"dato": "2026-09-15T09:00:00+02:00",
|
|
"beslutning": "ja",
|
|
}
|
|
record.update(overrides)
|
|
return record
|
|
|
|
|
|
def test_append_writes_one_line_and_leaves_prior_bytes_untouched(tmp_path):
|
|
path = str(tmp_path / "beslutninger.jsonl")
|
|
jsonl.append_line(path, entry(id="b1"))
|
|
before = open(path, "rb").read()
|
|
|
|
jsonl.append_line(path, entry(id="b2"))
|
|
after = open(path, "rb").read()
|
|
|
|
# Not "the file still parses" -- the earlier bytes must be the same bytes.
|
|
assert after.startswith(before)
|
|
assert after[len(before):].count(b"\n") == 1
|
|
assert [rec["id"] for rec in jsonl.read_lines(path)] == ["b1", "b2"]
|
|
|
|
|
|
def test_two_concurrent_writers_produce_two_intact_lines(tmp_path):
|
|
path = str(tmp_path / "logg.jsonl")
|
|
start = threading.Barrier(2)
|
|
count = 25
|
|
|
|
def writer(tag):
|
|
start.wait()
|
|
for index in range(count):
|
|
jsonl.append_line(path, entry(id="%s-%d" % (tag, index), notat="x" * 180))
|
|
|
|
threads = [threading.Thread(target=writer, args=(tag,)) for tag in ("a", "b")]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
records = jsonl.read_lines(path)
|
|
assert len(records) == 2 * count
|
|
assert len({rec["id"] for rec in records}) == 2 * count
|
|
|
|
|
|
def test_a_torn_trailing_line_is_skipped_not_fatal(tmp_path):
|
|
path = str(tmp_path / "logg.jsonl")
|
|
jsonl.append_line(path, entry(id="b1"))
|
|
with open(path, "a", encoding="utf-8") as handle:
|
|
handle.write('{"type": "beslutning", "id": "b2", "dat') # killed mid-write
|
|
|
|
records = jsonl.read_lines(path)
|
|
assert [rec["id"] for rec in records] == ["b1"]
|
|
|
|
|
|
def test_a_malformed_interior_line_raises_with_its_line_number(tmp_path):
|
|
path = str(tmp_path / "logg.jsonl")
|
|
jsonl.append_line(path, entry(id="b1"))
|
|
with open(path, "a", encoding="utf-8") as handle:
|
|
handle.write("{ not json at all }\n")
|
|
jsonl.append_line(path, entry(id="b3"))
|
|
|
|
with pytest.raises(jsonl.JsonlError) as caught:
|
|
jsonl.read_lines(path)
|
|
assert caught.value.line == 2
|
|
|
|
|
|
def test_fold_applies_a_correction_and_drops_an_outcome(tmp_path):
|
|
lines = [
|
|
entry(id="b1", beslutning="ja", arsak=["lonn"]),
|
|
entry(id="b2", beslutning="ja"),
|
|
entry(type="korrigering", id="k1", korrigerer="b1", beslutning="nei"),
|
|
entry(type="utfall", id="u1", korrigerer="b2", stadium="screening"),
|
|
]
|
|
folded = jsonl.fold(lines)
|
|
|
|
assert [rec["id"] for rec in folded] == ["b1", "b2"]
|
|
# The correction overwrote the decision it names, and did not invent a row.
|
|
assert folded[0]["beslutning"] == "nei"
|
|
assert folded[0]["arsak"] == ["lonn"]
|
|
# The outcome is not part of the decision stream; §5.7 mines it separately.
|
|
assert all(rec["type"] == "beslutning" for rec in folded)
|
|
|
|
|
|
def test_a_naive_timestamp_is_rejected_on_write(tmp_path):
|
|
path = str(tmp_path / "logg.jsonl")
|
|
with pytest.raises(jsonl.JsonlError):
|
|
jsonl.append_line(path, entry(dato="2026-09-15T09:00:00"))
|
|
# And nothing was written: a rejected record must not land half-way.
|
|
assert not os.path.exists(path) or open(path).read() == ""
|
|
|
|
|
|
def test_timestamps_across_the_dst_boundary_sort_by_instant_not_by_string():
|
|
before_shift = entry(id="before", dato="2026-10-25T02:30:00+02:00")
|
|
after_shift = entry(id="after", dato="2026-10-25T02:30:00+01:00")
|
|
|
|
# Same wall clock, an hour apart. String order is alphabetical on the
|
|
# offset and happens to be wrong here.
|
|
assert sorted([after_shift["dato"], before_shift["dato"]])[0] == after_shift["dato"]
|
|
assert [rec["id"] for rec in jsonl.by_time([after_shift, before_shift])] == [
|
|
"before",
|
|
"after",
|
|
]
|