open/jobbsok
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity Actions

feat(m2): add deterministic sak_status cli with golden output

Browse source 
--workspace is required with no environment fall-back: this process runs
unsandboxed, and a status run that quietly picked a workspace is one nobody
can audit. --check exits 1 on a stale cache, distinct from 2 for a failed run.

Byte-stability is measured, not asserted: the CLI is run twice under two
PYTHONHASHSEED values and the bytes compared, in both output formats. The
write boundary gets two digest tests rather than one -- without --oppdater
every byte stays put, with it exactly the diverging case's four cached keys
move and logg.jsonl, the body and every other file are byte-identical.

Co-Authored-By: Claude <claude-opus-5>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-05 21:40:32 +02:00
commit 12125c1932
3 changed files with 420 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

116
scripts/sak_status.py
View file

@ -403,3 +403,119 @@ def _dato(value):
def _iso(value):
return None if value is None else value.isoformat()
# ---------------------------------------------------------------------------
# Command line (plan Step 20)
#
# Output ordering is fully determined: cases come out sorted by sak-id, keys
# in JSON are sorted, and flags come out in FLAGG order. Nothing here iterates
# a dict or a set into stdout, which is what makes two runs byte-identical
# under different hash seeds rather than only usually.
# ---------------------------------------------------------------------------
#: Exit codes. 1 is reserved for `--check` finding divergence, so a caller can
#: tell "the cache is stale" apart from "the run failed".
EXIT_OK = 0
EXIT_DIVERGENS = 1
EXIT_FEIL = 2
def build_parser():
import argparse
parser = argparse.ArgumentParser(
prog="sak_status.py",
description=(
"Utled status, ventende part og stillhets-flagg for hver sak i "
"arbeidsomraadet. Leser logg.jsonl; sak.md er en hurtigbuffer."
),
)
# Required, and deliberately without the environment fall-backs
# paths.workspace_root offers: this process runs unsandboxed, and a
# status run that quietly picked a workspace would be a status run
# nobody could audit.
parser.add_argument("--workspace", required=True,
help="arbeidsomraadets rot (ingen underforstaatt standard)")
parser.add_argument("--today", default=None, metavar="YYYY-MM-DD",
help="klokka som skal brukes; injisert, aldri veggklokka")
parser.add_argument("--sak", default=None, metavar="SAK-ID",
help="begrens til en enkelt sak")
parser.add_argument("--format", dest="format", default="tekst",
choices=("tekst", "json"))
parser.add_argument("--check", action="store_true",
help="avslutt med kode 1 hvis en hurtigbuffer avviker")
parser.add_argument("--oppdater", action="store_true",
help="skriv de fire hurtigbuffer-noklene tilbake til sak.md")
return parser
def _skriv(strom, tekst):
strom.write(tekst)
strom.flush()
def _som_json(data):
import json as _json
return _json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
def _som_tekst(data):
linjer = ["sak-status %s (%d saker)" % (data["dato"], len(data["saker"])), ""]
for sak in data["saker"]:
linjer.append(sak["sak_id"])
linjer.append(" status: %-9s ventende: %-6s sist aktivitet: %s"
% (sak["status"], sak["ventende_part"],
sak["sist_aktivitet"] or "-"))
linjer.append(" neste frist: %s" % (sak["neste_frist"] or "-"))
if sak["flagg"]:
linjer.append(" flagg: %s" % ", ".join(sak["flagg"]))
if not sak["innkommende"]["registrert"]:
linjer.append(" innkommende: %s" % sak["innkommende"]["merknad"])
for avvik in sak["divergens"]:
linjer.append(" avvik: %s hurtigbuffer=%r utledet=%r"
% (avvik["nokkel"], avvik["hurtigbuffer"], avvik["utledet"]))
linjer.append("")
return "\n".join(linjer)
def main(argv=None, stdout=None, stderr=None):
import sys as _sys
argv = _sys.argv[1:] if argv is None else argv
stdout = _sys.stdout if stdout is None else stdout
stderr = _sys.stderr if stderr is None else stderr
args = build_parser().parse_args(argv)
today = args.today or datetime.date.today().isoformat()
try:
root = paths.workspace_root(args.workspace)
if args.sak is not None:
paths.validate_sak_id(args.sak)
data = rapport(root, today, args.sak)
if args.oppdater:
for sak in data["saker"]:
oppdater(root, sak["sak_id"], sak)
data = rapport(root, today, args.sak)
except (StatusError, paths.WorkspaceError, jsonl.JsonlError,
frontmatter_lib.FrontmatterError, OSError) as feil:
_skriv(stderr, "sak_status: %s\n" % feil)
return EXIT_FEIL
_skriv(stdout, _som_json(data) if args.format == "json" else _som_tekst(data))
if args.check:
avvikende = [sak for sak in data["saker"] if sak["divergens"]]
if avvikende:
_skriv(stdout, "\n%d sak(er) med utdatert hurtigbuffer. Kjor --oppdater.\n"
% len(avvikende))
return EXIT_DIVERGENS
return EXIT_OK
if __name__ == "__main__":
import sys as _sys
_sys.exit(main())

150
tests/golden/sak-status-2026-09-15.json Normal file
View file

@ -0,0 +1,150 @@
{
"dato": "2026-09-15",
"saker": [
{
"arbeidsgiver": "Aurland Maritim AS",
"divergens": [],
"flagg": [],
"innkommende": {
"merknad": "ingen innkommende hendelse er registrert. Med e-postserveren utsatt betyr det ikke at ingen finnes.",
"registrert": false,
"sist": null
},
"neste_frist": null,
"rolle": "Prosjektleder",
"sak_id": "2026-09-aurland-maritim-as-prosjektleder",
"sist_aktivitet": "2026-09-05",
"status": "trukket",
"ventende_part": "ingen"
},
{
"arbeidsgiver": "Bratthaug Industri AS",
"divergens": [],
"flagg": [
"intervju_10"
],
"innkommende": {
"merknad": null,
"registrert": true,
"sist": "2026-08-16"
},
"neste_frist": "2026-09-05",
"rolle": "Systemarkitekt",
"sak_id": "2026-09-bratthaug-industri-as-systemarkitekt",
"sist_aktivitet": "2026-08-26",
"status": "intervju",
"ventende_part": "dem"
},
{
"arbeidsgiver": "Havbris Energi AS",
"divergens": [
{
"hurtigbuffer": "dialog",
"nokkel": "status",
"utledet": "sendt"
},
{
"hurtigbuffer": "meg",
"nokkel": "ventende_part",
"utledet": "dem"
}
],
"flagg": [
"sendt_14"
],
"innkommende": {
"merknad": "ingen innkommende hendelse er registrert. Med e-postserveren utsatt betyr det ikke at ingen finnes.",
"registrert": false,
"sist": null
},
"neste_frist": "2026-09-09",
"rolle": "Løsningsarkitekt",
"sak_id": "2026-09-havbris-energi-as-losningsarkitekt",
"sist_aktivitet": "2026-08-27",
"status": "sendt",
"ventende_part": "dem"
},
{
"arbeidsgiver": "Myrvang Teknologi AS",
"divergens": [],
"flagg": [],
"innkommende": {
"merknad": "ingen innkommende hendelse er registrert. Med e-postserveren utsatt betyr det ikke at ingen finnes.",
"registrert": false,
"sist": null
},
"neste_frist": null,
"rolle": "Utvikler",
"sak_id": "2026-09-myrvang-teknologi-as-utvikler",
"sist_aktivitet": "2026-08-06",
"status": "avslag",
"ventende_part": "ingen"
},
{
"arbeidsgiver": "Nordlys Data AS",
"divergens": [],
"flagg": [],
"innkommende": {
"merknad": "ingen innkommende hendelse er registrert. Med e-postserveren utsatt betyr det ikke at ingen finnes.",
"registrert": false,
"sist": null
},
"neste_frist": null,
"rolle": "AI-rådgiver",
"sak_id": "2026-09-nordlys-data-as-ai-radgiver",
"sist_aktivitet": "2026-09-13",
"status": "vurderer",
"ventende_part": "meg"
},
{
"arbeidsgiver": "Solvang Helse AS",
"divergens": [],
"flagg": [],
"innkommende": {
"merknad": null,
"registrert": true,
"sist": "2026-09-12"
},
"neste_frist": null,
"rolle": "Teknisk rådgiver",
"sak_id": "2026-09-solvang-helse-as-teknisk-radgiver",
"sist_aktivitet": "2026-09-12",
"status": "tilbud",
"ventende_part": "meg"
},
{
"arbeidsgiver": "Storelva kommune",
"divergens": [],
"flagg": [
"dialog_7"
],
"innkommende": {
"merknad": null,
"registrert": true,
"sist": "2026-08-21"
},
"neste_frist": "2026-08-28",
"rolle": "Fagleder digitalisering",
"sak_id": "2026-09-storelva-kommune-fagleder-digitalisering",
"sist_aktivitet": "2026-08-22",
"status": "dialog",
"ventende_part": "dem"
},
{
"arbeidsgiver": "Værøy Sjømat AS",
"divergens": [],
"flagg": [],
"innkommende": {
"merknad": "ingen innkommende hendelse er registrert. Med e-postserveren utsatt betyr det ikke at ingen finnes.",
"registrert": false,
"sist": null
},
"neste_frist": null,
"rolle": "Dataingeniør",
"sak_id": "2026-09-vaeroy-sjomat-as-dataingenior",
"sist_aktivitet": "2026-09-10",
"status": "soker",
"ventende_part": "meg"
}
]
}

154
tests/test_sak_status_cli.py Normal file
View file

@ -0,0 +1,154 @@
"""The sak_status command line, and what it is allowed to write (plan Step 20).
Two properties are asserted here that no unit test can reach, because both are
about the process rather than the function.
**Byte-stability.** The golden file is only evidence if the same workspace and
the same `--today` produce the same bytes every time, including under a
different `PYTHONHASHSEED`. Python randomises string hashing per process, so a
dictionary or a set that reached stdout would reorder between runs and the
golden would fail intermittently -- the worst way for a determinism bug to
present itself. Running the CLI twice under two seeds and comparing bytes is
what turns "no dict iteration order reaches stdout" from an intention into a
measurement.
**The write boundary.** The read paths write nothing, and `--oppdater` is the
one exception -- not a caveat, a named mode. So there are two digest tests,
not one: a run without the flag leaves every byte in the workspace where it
was, and a run with it changes exactly the four cached frontmatter keys and
leaves `logg.jsonl`, the `sak.md` body and every other file byte-identical.
Saying "the script writes nothing" without that second test is what let the
write-back go unnoticed in the first place.
Style note: this file follows tests/test_sak_status_silence.py.
"""
import hashlib
import json
import os
import subprocess
import sys
import pytest
import sak_status
from jobbsok_lib import frontmatter as frontmatter_lib
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CLI = os.path.join(REPO, "scripts", "sak_status.py")
KORPUS = os.path.join(REPO, "tests", "fixtures", "workspace")
GOLDEN = os.path.join(REPO, "tests", "golden", "sak-status-2026-09-15.json")
TODAY = "2026-09-15"
#: The case whose cached frontmatter is deliberately wrong (Step 19).
UENIG = "2026-09-havbris-energi-as-losningsarkitekt"
def kjor(*argv, **kwargs):
"""Run the CLI as a process and return (exit code, stdout, stderr) bytes."""
env = dict(os.environ)
env.pop("JOBBSOK_WORKSPACE", None)
env.pop("CLAUDE_PLUGIN_DATA", None)
env.update(kwargs.get("env", {}))
proc = subprocess.run(
[sys.executable, CLI] + list(argv),
stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env,
)
return proc.returncode, proc.stdout, proc.stderr
def digest(root):
"""Every file under ``root`` mapped to the sha256 of its bytes."""
ut = {}
for dirpath, dirnames, filenames in os.walk(root):
dirnames.sort()
for navn in sorted(filenames):
full = os.path.join(dirpath, navn)
with open(full, "rb") as handle:
ut[os.path.relpath(full, root)] = hashlib.sha256(handle.read()).hexdigest()
return ut
def test_the_json_output_matches_the_golden_file(golden):
kode, ut, feil = kjor("--workspace", KORPUS, "--today", TODAY, "--format", "json")
assert kode == 0, feil.decode("utf-8")
golden(GOLDEN, ut.decode("utf-8"))
def test_two_consecutive_runs_are_byte_identical():
_kode, forste, _ = kjor("--workspace", KORPUS, "--today", TODAY, "--format", "json")
_kode, andre, _ = kjor("--workspace", KORPUS, "--today", TODAY, "--format", "json")
assert forste == andre
def test_two_different_hash_seeds_produce_the_same_bytes():
kwargs = ("--workspace", KORPUS, "--today", TODAY, "--format", "json")
_kode, null, _ = kjor(*kwargs, env={"PYTHONHASHSEED": "0"})
_kode, mange, _ = kjor(*kwargs, env={"PYTHONHASHSEED": "1048573"})
assert null == mange, (
"output moved with PYTHONHASHSEED, so a dict or a set reached stdout"
)
# And the text form too, which is the one the operator actually reads.
tekst = ("--workspace", KORPUS, "--today", TODAY)
_kode, a, _ = kjor(*tekst, env={"PYTHONHASHSEED": "0"})
_kode, b, _ = kjor(*tekst, env={"PYTHONHASHSEED": "1048573"})
assert a == b
def test_omitting_workspace_is_an_error_and_never_a_guess_at_home():
kode, ut, feil = kjor("--today", TODAY)
assert kode != 0
assert b"--workspace" in feil
assert os.path.expanduser("~").encode("utf-8") not in ut + feil, (
"the refusal leaked a home directory; there is no default workspace"
)
def test_a_traversing_case_identifier_is_refused():
kode, _ut, feil = kjor(
"--workspace", KORPUS, "--today", TODAY, "--sak", "../../etc/passwd"
)
assert kode != 0
assert b"../../etc/passwd" in feil
def test_a_run_without_oppdater_leaves_every_byte_where_it_was(workspace):
forst = digest(workspace)
kode, _ut, feil = kjor("--workspace", workspace, "--today", TODAY, "--format", "json")
assert kode == 0, feil.decode("utf-8")
assert digest(workspace) == forst, "the read path wrote something"
def test_a_run_with_oppdater_changes_only_the_four_cached_keys(workspace):
sti = os.path.join(workspace, "saker", UENIG, "sak.md")
with open(sti, "r", encoding="utf-8") as handle:
for_meta, for_body = frontmatter_lib.parse(handle.read())
forst = digest(workspace)
kode, _ut, feil = kjor("--workspace", workspace, "--today", TODAY, "--oppdater")
assert kode == 0, feil.decode("utf-8")
etter = digest(workspace)
endrede = sorted(navn for navn in forst if forst[navn] != etter[navn])
assert endrede == [os.path.join("saker", UENIG, "sak.md")], (
"--oppdater touched %r; only the diverging case's sak.md may move"
% (endrede,)
)
assert sorted(etter) == sorted(forst), "--oppdater added or removed a file"
with open(sti, "r", encoding="utf-8") as handle:
etter_meta, etter_body = frontmatter_lib.parse(handle.read())
assert etter_body == for_body
flyttet = sorted(k for k in for_meta if for_meta[k] != etter_meta.get(k))
assert flyttet == ["status", "ventende_part"]
assert set(flyttet) <= set(sak_status.CACHE_KEYS)
def test_check_exits_non_zero_on_the_deliberately_divergent_fixture(workspace):
kode, ut, _feil = kjor("--workspace", KORPUS, "--today", TODAY, "--check")
assert kode != 0
assert UENIG.encode("utf-8") in ut, "the check must name the case that diverged"
kjor("--workspace", workspace, "--today", TODAY, "--oppdater")
kode, _ut, _feil = kjor("--workspace", workspace, "--today", TODAY, "--check")
assert kode == 0, "a workspace that has just been refreshed must check clean"