125 lines
4.7 KiB
Python
125 lines
4.7 KiB
Python
"""Workspace construction and corpus-integrity helpers (plan Step 6).
|
|
|
|
`make_case` builds the case shape from build-brief 5.2 and 5.3 so tests can
|
|
say what they need in one line instead of laying out directories by hand.
|
|
`snapshot_tree` and `assert_tree_unchanged` are the machinery behind
|
|
conftest's autouse pristine guard: a test that writes into the fixture corpus
|
|
poisons every later run, and the failure would otherwise surface far from its
|
|
cause.
|
|
|
|
Note on the case log: entries here carry `hendelse` from the build-brief 5.3
|
|
enum and are written with a single append, mirroring
|
|
`jobbsok_lib.jsonl.append_line`'s contract without going through its
|
|
`type` validation -- that closed set (`beslutning`, `korrigering`, `utfall`)
|
|
is the decision log's discriminator, and the case log has its own. Step 17
|
|
defines the event enum properly; until then this helper does not force a
|
|
decision-log type onto a case event. Reading goes back through
|
|
`jsonl.read_lines`, so the torn-line and corruption rules still apply.
|
|
"""
|
|
|
|
import datetime
|
|
import json
|
|
import os
|
|
|
|
from jobbsok_lib import frontmatter as frontmatter_lib
|
|
from jobbsok_lib import jsonl, paths
|
|
|
|
#: The frozen date the whole suite reasons from, as an aware instant. Norway
|
|
#: is on CEST (+02:00) on this date, and the offset is written out rather than
|
|
#: derived, so the fixtures do not depend on the machine's timezone database.
|
|
FROZEN_TODAY = "2026-09-15"
|
|
FROZEN_NOON = datetime.datetime.fromisoformat(FROZEN_TODAY + "T12:00:00+02:00")
|
|
|
|
|
|
def event(hendelse, days_ago=0, kilde=None, ref=None, notat=None):
|
|
"""A build-brief 5.3 case-log record, dated relative to the frozen today."""
|
|
moment = FROZEN_NOON - datetime.timedelta(days=days_ago)
|
|
return {
|
|
"ts": moment.isoformat(),
|
|
"hendelse": hendelse,
|
|
"kilde": kilde,
|
|
"ref": ref,
|
|
"notat": notat,
|
|
}
|
|
|
|
|
|
def make_case(root, arbeidsgiver, rolle, year_month, events=(), **fields):
|
|
"""Create ``saker/<sak-id>/`` with a sak.md and a logg.jsonl.
|
|
|
|
Returns the `sak-id`, which is derived through :func:`paths.sak_id` rather
|
|
than assembled here -- the slug rules are that module's contract and a
|
|
second implementation would drift from it.
|
|
"""
|
|
sak_id = paths.sak_id(year_month, arbeidsgiver, rolle)
|
|
case_dir = paths.safe_join(root, "saker", sak_id)
|
|
os.makedirs(case_dir, exist_ok=True)
|
|
|
|
metadata = {
|
|
"sak_id": sak_id,
|
|
"arbeidsgiver": arbeidsgiver,
|
|
"rolle": rolle,
|
|
"opprettet": FROZEN_TODAY,
|
|
"status": "vurderes",
|
|
"ventende_part": "meg",
|
|
}
|
|
metadata.update(fields)
|
|
with open(os.path.join(case_dir, "sak.md"), "w", encoding="utf-8") as handle:
|
|
handle.write(frontmatter_lib.render(metadata, "## Hvorfor denne\n\n"))
|
|
|
|
log = os.path.join(case_dir, "logg.jsonl")
|
|
for record in events:
|
|
_append_raw(log, record)
|
|
if not events:
|
|
open(log, "a", encoding="utf-8").close()
|
|
return sak_id
|
|
|
|
|
|
def read_jsonl(path):
|
|
"""Read a JSONL file through the real reader, torn-line rules and all."""
|
|
return jsonl.read_lines(path)
|
|
|
|
|
|
def frontmatter(path):
|
|
"""Parse a file's frontmatter, returning ``(metadata, body)``."""
|
|
with open(path, "r", encoding="utf-8") as handle:
|
|
return frontmatter_lib.parse(handle.read())
|
|
|
|
|
|
def snapshot_tree(root):
|
|
"""Map every file under ``root`` to the pair that betrays a mutation.
|
|
|
|
Size catches an edit that changes length, modification time catches one
|
|
that does not. Neither alone is enough, and a content hash would make the
|
|
autouse guard cost real time on every test in the suite.
|
|
"""
|
|
snapshot = {}
|
|
for dirpath, _dirnames, filenames in os.walk(root):
|
|
for name in filenames:
|
|
full = os.path.join(dirpath, name)
|
|
info = os.stat(full)
|
|
snapshot[os.path.relpath(full, root)] = (info.st_size, info.st_mtime_ns)
|
|
return snapshot
|
|
|
|
|
|
def assert_tree_unchanged(root, before):
|
|
"""Fail if anything under ``root`` was added, removed or modified."""
|
|
after = snapshot_tree(root)
|
|
added = sorted(set(after) - set(before))
|
|
removed = sorted(set(before) - set(after))
|
|
changed = sorted(name for name in set(before) & set(after) if before[name] != after[name])
|
|
if not (added or removed or changed):
|
|
return
|
|
raise AssertionError(
|
|
"the fixture corpus under %r was mutated by a test -- added=%r "
|
|
"removed=%r changed=%r. Fixtures are read-only; copy into tmp_path "
|
|
"instead." % (root, added, removed, changed)
|
|
)
|
|
|
|
|
|
def _append_raw(path, record):
|
|
payload = (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8")
|
|
handle = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
|
|
try:
|
|
os.write(handle, payload)
|
|
finally:
|
|
os.close(handle)
|