163 lines
6.8 KiB
Python
163 lines
6.8 KiB
Python
"""The append-only sourcing-decision store (plan Step 22).
|
|
|
|
Build-brief 5.4 is a schema, and this file is what keeps it one. The store
|
|
cannot be migrated -- that is the whole consequence of append-only (risk H10) --
|
|
so every rule it will ever have has to hold on the first line written, and the
|
|
fixture corpus under `tests/fixtures/beslutninger/` is the fasit the writer is
|
|
measured against rather than a sample it may drift from.
|
|
|
|
Four of the six rules below exist because the field they guard is read much
|
|
later by something that cannot ask again:
|
|
|
|
* **`score_da` is an integer.** M6 reads it as a value on a weight-dependent
|
|
scale (risk M8r). A string that looks like a number sorts as text and
|
|
averages as nothing.
|
|
* **One to three reasons, from the closed twelve.** `laering` proposes only on
|
|
"the same `arsak` in at least four of twenty". A thirteenth reason invented
|
|
at write time is a category that can never reach that threshold and never be
|
|
counted against it either.
|
|
* **A note is at most fifteen words.** The cap is what keeps the field a label
|
|
rather than a second body, and build-brief 5.4 states it as a number.
|
|
* **A correction names a decision that exists.** `jsonl.fold` raises when it
|
|
meets a correction pointing at nothing; catching it at write time is the
|
|
difference between a refusal and a log that cannot be read afterwards.
|
|
|
|
Style note: this file follows tests/test_jsonl.py.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
|
|
import pytest
|
|
|
|
import beslutninger
|
|
from jobbsok_lib import jsonl
|
|
|
|
|
|
def beslutning(**overrides):
|
|
record = {
|
|
"id": "b-0001",
|
|
"dato": "2026-09-01T09:00:00+02:00",
|
|
"kilde": "manuell",
|
|
"url": "https://stillinger.example/sak-0001",
|
|
"tittel": "Dataingeniør",
|
|
"arbeidsgiver": "Værøy Sjømat AS",
|
|
"beslutning": "ja",
|
|
"arsak": ["fagomrade", "teknologi"],
|
|
"notat": "treffer kjernen",
|
|
"score_da": 78,
|
|
"delscore": {"fagomrade": 78, "oppgavetype": 73, "teknologi": 68, "selskapstype": 63},
|
|
"vekter": {"fagomrade": 40, "oppgavetype": 30, "teknologi": 20, "selskapstype": 10},
|
|
"vekt_hash": "sha256:" + "e4" * 32,
|
|
}
|
|
record.update(overrides)
|
|
return record
|
|
|
|
|
|
def test_an_append_adds_one_line_and_leaves_prior_bytes_unchanged(empty_workspace):
|
|
sti = beslutninger.sti(empty_workspace)
|
|
beslutninger.legg_til(empty_workspace, beslutning(id="b-0001"))
|
|
with open(sti, "rb") as handle:
|
|
forst = handle.read()
|
|
|
|
beslutninger.legg_til(empty_workspace, beslutning(id="b-0002", beslutning="nei",
|
|
arsak=["lonn"]))
|
|
with open(sti, "rb") as handle:
|
|
etter = handle.read()
|
|
|
|
# Not "the file still parses" -- the earlier bytes must be the same bytes.
|
|
assert etter.startswith(forst)
|
|
assert etter[len(forst):].count(b"\n") == 1
|
|
assert [rec["id"] for rec in beslutninger.les(empty_workspace)] == ["b-0001", "b-0002"]
|
|
|
|
# And the line carries the two fields no later migration can add.
|
|
forste = json.loads(forst.decode("utf-8"))
|
|
assert forste["type"] == "beslutning" and forste["skjema"] == jsonl.SKJEMA
|
|
assert list(forste) == list(beslutninger.FELTREKKEFOLGE), (
|
|
"the record is laid out as %r; the fixture corpus is the fasit and it "
|
|
"lays it out as %r" % (list(forste), list(beslutninger.FELTREKKEFOLGE))
|
|
)
|
|
|
|
|
|
def test_one_to_three_reasons_from_the_enum_are_accepted_and_nothing_else_is(
|
|
empty_workspace,
|
|
):
|
|
for arsak in (["lonn"], ["lonn", "geografi"], ["lonn", "geografi", "annet"]):
|
|
record = beslutninger.legg_til(
|
|
empty_workspace,
|
|
beslutning(id="b-%d" % len(arsak) + "-".join(arsak), arsak=arsak),
|
|
)
|
|
assert record["arsak"] == arsak
|
|
|
|
for ugyldig in ([], ["lonn", "geografi", "annet", "tidspunkt"], ["pendling"],
|
|
["lonn", "lonn"], "lonn"):
|
|
with pytest.raises(beslutninger.BeslutningError) as fanget:
|
|
beslutninger.legg_til(empty_workspace, beslutning(id="b-avvist", arsak=ugyldig))
|
|
assert "arsak" in str(fanget.value)
|
|
|
|
# The refusal leaves nothing behind: validation runs before the file opens.
|
|
assert [rec["id"] for rec in beslutninger.les(empty_workspace)] == [
|
|
"b-1lonn", "b-2lonn-geografi", "b-3lonn-geografi-annet"
|
|
]
|
|
|
|
|
|
def test_a_note_longer_than_fifteen_words_is_refused(empty_workspace):
|
|
akkurat = " ".join("ord%d" % n for n in range(beslutninger.MAKS_NOTAT_ORD))
|
|
record = beslutninger.legg_til(empty_workspace, beslutning(notat=akkurat))
|
|
assert record["notat"] == akkurat
|
|
|
|
ett_for_mye = akkurat + " overflodig"
|
|
with pytest.raises(beslutninger.BeslutningError) as fanget:
|
|
beslutninger.legg_til(empty_workspace, beslutning(id="b-lang", notat=ett_for_mye))
|
|
assert "15" in str(fanget.value), (
|
|
"the refusal must name the cap it enforced: %r" % (str(fanget.value),)
|
|
)
|
|
|
|
|
|
def test_score_da_must_be_an_integer(empty_workspace):
|
|
for ugyldig in ("78", 78.5, None, True):
|
|
with pytest.raises(beslutninger.BeslutningError) as fanget:
|
|
beslutninger.legg_til(
|
|
empty_workspace, beslutning(id="b-score", score_da=ugyldig)
|
|
)
|
|
assert "score_da" in str(fanget.value)
|
|
|
|
record = beslutninger.legg_til(empty_workspace, beslutning(score_da=0))
|
|
assert record["score_da"] == 0 and isinstance(record["score_da"], int)
|
|
|
|
|
|
def test_a_correction_must_reference_a_decision_that_exists(empty_workspace):
|
|
beslutninger.legg_til(empty_workspace, beslutning(id="b-0001", arsak=["lonn"]))
|
|
|
|
with pytest.raises(beslutninger.BeslutningError) as fanget:
|
|
beslutninger.korriger(
|
|
empty_workspace,
|
|
{"id": "k-0001", "dato": "2026-09-02T09:00:00+02:00",
|
|
"korrigerer": "b-finnes-ikke", "arsak": ["geografi"]},
|
|
)
|
|
assert "b-finnes-ikke" in str(fanget.value)
|
|
|
|
rettelse = beslutninger.korriger(
|
|
empty_workspace,
|
|
{"id": "k-0001", "dato": "2026-09-02T09:00:00+02:00",
|
|
"korrigerer": "b-0001", "arsak": ["geografi"],
|
|
"notat": "reisevei var den egentlige grunnen"},
|
|
)
|
|
assert rettelse["type"] == "korrigering"
|
|
|
|
# A correction is a new line, never an edit: both are still on disk, and
|
|
# the folded stream shows the corrected decision rather than two rows.
|
|
linjer = beslutninger.les(empty_workspace)
|
|
assert [rec["id"] for rec in linjer] == ["b-0001", "k-0001"]
|
|
strom = beslutninger.stromme(empty_workspace)
|
|
assert len(strom) == 1
|
|
assert strom[0]["id"] == "b-0001" and strom[0]["arsak"] == ["geografi"]
|
|
|
|
|
|
def test_the_case_identifier_from_a_norwegian_employer_is_pure_ascii():
|
|
sak_id = beslutninger.sak_id_for(beslutning())
|
|
assert sak_id == "2026-09-vaeroy-sjomat-as-dataingenior"
|
|
sak_id.encode("ascii") # raises if a macOS NFD byte survived (risk H8)
|
|
|
|
# And the hand-off only happens on a yes: a `nei` creates no case at all.
|
|
assert beslutninger.sak_id_for(beslutning(beslutning="nei")) is None
|