Kjell Tore Guttormsen
9847e014e7
The eight U-IDs and their type pointers were checked against the approved list (no deviation) and the data file now says approved, with the date and the source. Row 5 no longer reports "not approved" but the measured count: 3 of 8 (U12, U4, U6), because a point counts only when every type it points at is green and only types 1, 3 and 7 are. No gate logic changed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
431 lines
17 KiB
Python
431 lines
17 KiB
Python
"""The v1 gate's own tests: every row CAN go green and CAN go red.
|
||
|
||
A gate that can only be red is as worthless as one that can only be green, so each row is driven
|
||
from fixtures on both sides of its line. The probes and the stress measurement are injected here
|
||
(``probe_runner`` / ``stress_measure``) so the logic is exercised without a child pytest; one
|
||
subprocess arm runs the real command end to end.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from portfolio_optimiser.evals import v1_gate as gate
|
||
|
||
_REPO = Path(__file__).resolve().parents[1]
|
||
_CONFIG = gate.load_config()
|
||
_AI = gate.ai_authored_lines(_REPO, _CONFIG["ai_authored"])
|
||
_ALL_NODEIDS = [n for spec in _CONFIG["feedback_types"].values() for n in spec["evidence"]] + list(
|
||
_CONFIG["row6_evidence"]
|
||
)
|
||
|
||
|
||
def _write(path: Path, payload: Any) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
text = payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False)
|
||
path.write_text(text, encoding="utf-8")
|
||
|
||
|
||
def _feedback(round_dir: Path, *items: tuple[str, int, str], author: str = "fagperson") -> None:
|
||
_write(
|
||
round_dir / "feedback.json",
|
||
{"author": author, "items": [{"id": i, "type": t, "text": x} for i, t, x in items]},
|
||
)
|
||
|
||
|
||
def _outcome(round_dir: Path, rows: list[dict[str, Any]], removed: Any = ()) -> None:
|
||
_write(
|
||
round_dir / "outcome.json",
|
||
{"run_id": f"r{round_dir.name}", "approaches": rows, "removed": list(removed)},
|
||
)
|
||
|
||
|
||
def _row(aid: str, validated: bool, nok: float | None, *ids: str, stage: str = "") -> dict:
|
||
return {
|
||
"id": aid,
|
||
"validated": validated,
|
||
"stage": stage,
|
||
"validated_nok": nok,
|
||
"feedback_ids": list(ids),
|
||
}
|
||
|
||
|
||
def _green_rounds(root: Path) -> Path:
|
||
"""Three traced rounds and a round 3 report kept at 100 %."""
|
||
_outcome(root / "0", [_row("a1", False, None, stage="stage0")])
|
||
for n in (1, 2, 3):
|
||
_feedback(root / str(n), (f"f{n}", 1, f"Tallet for linje {n} er feil, bruk kontrakten."))
|
||
_outcome(root / str(n), [_row("a1", True, 1000.0 * n, f"f{n}")])
|
||
_write(root / str(n) / "report.md", f"# Rapport {n}\n\nlinje\n")
|
||
_write(root / "3" / "report.kept.md", "# Rapport 3\n\nlinje\n")
|
||
return root
|
||
|
||
|
||
def _all_pass(ids: Any) -> dict[str, str]:
|
||
return {n: "passed" for n in ids}
|
||
|
||
|
||
_CLEAN = gate.StressMeasure(
|
||
validated=10, undeclared=0, named=1, rows=20, commissioned=20, where="x"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Row 1 — rounds with a real domain expert
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
def test_row1_is_red_with_no_rounds_and_green_with_three(tmp_path: Path) -> None:
|
||
assert gate.score_rounds(tmp_path / "none", 3, _AI).k == 0
|
||
assert gate.score_rounds(tmp_path / "none", 3, _AI).status == gate.RED
|
||
row = gate.score_rounds(_green_rounds(tmp_path / "r"), 3, _AI)
|
||
assert (row.k, row.status) == (3, gate.GREEN)
|
||
|
||
|
||
def test_row1_an_empty_or_wrongly_shaped_feedback_does_not_count(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_write(root / "1" / "feedback.json", {"author": "fagperson", "items": []})
|
||
(root / "2" / "feedback.json").unlink()
|
||
_write(root / "2" / "feedback.md", "Dette er min tilbakemelding.")
|
||
_feedback(root / "3", ("f3", 1, "noe"), author="")
|
||
row = gate.score_rounds(root, 3, _AI)
|
||
assert row.k == 0
|
||
assert any("feedback.md" in x for x in row.exceptions)
|
||
|
||
|
||
def test_row1_the_ai_authored_answer_sheet_can_never_be_counted_in(tmp_path: Path) -> None:
|
||
"""``docs/ekspert-svar.md`` is AI-authored: text lifted from it is refused, and a control with
|
||
the expert's own words in the same shape IS counted."""
|
||
doc = (_REPO / "docs" / "ekspert-svar.md").read_text(encoding="utf-8")
|
||
lifted = next(line for line in doc.splitlines() if "Skal en dom telle som fagdom" in line)
|
||
root = _green_rounds(tmp_path)
|
||
_feedback(root / "1", ("f1", 1, lifted.lstrip("> ")))
|
||
_feedback(root / "2", ("f2", 1, "Se her: " + lifted.lstrip("> ") + " Takk."))
|
||
row = gate.score_rounds(root, 3, _AI)
|
||
assert row.k == 1
|
||
assert sum("AI-forfattet" in x for x in row.exceptions) == 2
|
||
# The guard fails closed when it cannot read its source.
|
||
assert gate.score_rounds(root, 3, None).k == 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Row 2 — rounds with a measurable change
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
def test_row2_a_traced_change_counts(tmp_path: Path) -> None:
|
||
row = gate.score_changes(_green_rounds(tmp_path), 3, _AI)
|
||
assert (row.k, row.status) == (3, gate.GREEN)
|
||
assert "runde 0 =" in row.reason and "(r0)" in row.reason
|
||
|
||
|
||
def test_row2_a_change_without_a_trace_is_model_noise(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_outcome(root / "2", [_row("a1", True, 2000.0)]) # changed nok, no feedback id
|
||
_outcome(root / "3", [_row("a1", True, 3000.0, "f1")]) # traced to an EARLIER round's id
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert (row.k, row.status) == (1, gate.RED)
|
||
assert sum("ingen sporet" in x for x in row.exceptions) == 2
|
||
|
||
|
||
def test_row2_no_change_does_not_count(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_outcome(root / "2", [_row("a1", True, 1000.0, "f2")]) # identical to round 1
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert row.k == 2
|
||
assert any("ingen endring" in x for x in row.exceptions)
|
||
|
||
|
||
def test_row2_each_of_a_to_d_is_a_change(tmp_path: Path) -> None:
|
||
ids = {"f"}
|
||
base = [_row("a1", False, None, stage="stage0")]
|
||
cases = {
|
||
"a-added": [*base, _row("a2", False, None, "f", stage="stage0")],
|
||
"b-validated": [_row("a1", True, None, "f", stage="stage0")],
|
||
"c-stage": [_row("a1", False, None, "f", stage="stage4")],
|
||
"d-nok": [_row("a1", False, 5.0, "f", stage="stage0")],
|
||
}
|
||
for name, rows in cases.items():
|
||
_outcome(tmp_path / name / "0", base)
|
||
_outcome(tmp_path / name / "1", rows)
|
||
ok, why = gate.round_changed(
|
||
tmp_path / name / "0" / "outcome.json", tmp_path / name / "1" / "outcome.json", ids
|
||
)
|
||
assert ok, (name, why)
|
||
# (a) by removal: traced only through the ``removed`` list.
|
||
_outcome(tmp_path / "rm" / "0", [*base, _row("a2", False, None)])
|
||
_outcome(tmp_path / "rm" / "1", base)
|
||
assert not gate.round_changed(
|
||
tmp_path / "rm" / "0" / "outcome.json", tmp_path / "rm" / "1" / "outcome.json", ids
|
||
)[0]
|
||
_outcome(tmp_path / "rm" / "1", base, removed=[{"id": "a2", "feedback_ids": ["f"]}])
|
||
assert gate.round_changed(
|
||
tmp_path / "rm" / "0" / "outcome.json", tmp_path / "rm" / "1" / "outcome.json", ids
|
||
)[0]
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Row 3 — feedback types
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
def test_row3_all_probes_passing_is_eight_of_eight() -> None:
|
||
row = gate.score_types(_CONFIG["feedback_types"], _all_pass(_ALL_NODEIDS))
|
||
assert (row.k, row.n, row.status) == (8, 8, gate.GREEN)
|
||
|
||
|
||
def test_row3_partial_is_no_and_a_missing_probe_is_red() -> None:
|
||
types = json.loads(json.dumps(_CONFIG["feedback_types"]))
|
||
outcomes = _all_pass(_ALL_NODEIDS)
|
||
outcomes[types["7"]["evidence"][1]] = "failed" # one of type 7's two tests
|
||
types["4"]["evidence"] = []
|
||
outcomes[types["1"]["evidence"][0]] = "missing"
|
||
row = gate.score_types(types, outcomes)
|
||
assert (row.k, row.status) == (5, gate.RED)
|
||
assert {x.split()[1] for x in row.exceptions} == {"1", "4", "7"}
|
||
|
||
|
||
def test_row3_every_registered_test_exists() -> None:
|
||
"""A renamed test would silently turn a type red; this names the drift instead."""
|
||
for nodeid in _ALL_NODEIDS:
|
||
path, name = nodeid.split("::")
|
||
tree = ast.parse((_REPO / path).read_text(encoding="utf-8"))
|
||
names = {n.name for n in tree.body if isinstance(n, ast.FunctionDef | ast.AsyncFunctionDef)}
|
||
assert name in names, nodeid
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Row 4 — round 3 kept
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize(("kept_lines", "status"), [(79, gate.RED), (80, gate.GREEN)])
|
||
def test_row4_the_line_is_eighty_percent(tmp_path: Path, kept_lines: int, status: str) -> None:
|
||
lines = [f"linje {i}" for i in range(100)]
|
||
_write(tmp_path / "3" / "report.md", "\n\n".join(lines) + "\n")
|
||
kept = lines[:kept_lines] + [f"endret {i}" for i in range(100 - kept_lines)]
|
||
_write(tmp_path / "3" / "report.kept.md", "\n".join(kept) + "\n")
|
||
row = gate.score_kept(tmp_path, 0.8)
|
||
assert (row.k, row.n, row.status) == (kept_lines, 100, status)
|
||
|
||
|
||
def test_row4_a_missing_kept_report_is_red_never_full(tmp_path: Path) -> None:
|
||
_write(tmp_path / "3" / "report.md", "a\nb\n")
|
||
row = gate.score_kept(tmp_path, 0.8)
|
||
assert (row.k, row.status, row.reason) == (None, gate.RED, "ingen rapport")
|
||
|
||
|
||
def test_row4_a_line_kept_once_counts_once(tmp_path: Path) -> None:
|
||
_write(tmp_path / "3" / "report.md", "x\nx\ny\n")
|
||
_write(tmp_path / "3" / "report.kept.md", "x\nz\n")
|
||
assert gate.score_kept(tmp_path, 0.8).k == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Row 5 — MAF points
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
def test_row5_is_red_until_the_operator_approves_the_list() -> None:
|
||
unapproved = {**_CONFIG["maf_points"], "approved": False}
|
||
verdicts = gate.green_types(_CONFIG["feedback_types"], _all_pass(_ALL_NODEIDS))
|
||
row = gate.score_maf(unapproved, verdicts, gate._PACKAGE_SRC)
|
||
assert (row.k, row.status, row.reason) == (0, gate.RED, "M ikke godkjent av operatøren")
|
||
assert "presence 7 av 8" in row.diagnostics
|
||
|
||
|
||
def test_row5_the_approved_list_counts_only_points_whose_types_are_green() -> None:
|
||
"""The tracked list is operator-approved (17.09): eight U-IDs, and a point counts only when
|
||
every type it points at is green. With today's green types (1, 3, 7) that is U12, U4, U6."""
|
||
maf = _CONFIG["maf_points"]
|
||
assert (maf["approved"], maf["approved_on"], maf["approved_by"]) == (
|
||
True,
|
||
"2026-09-17",
|
||
"operatørgodkjent",
|
||
)
|
||
assert [(p["u_id"], p["types"]) for p in maf["points"]] == [
|
||
("U13", [1, 2]),
|
||
("U9", [1, 8]),
|
||
("U12", [1]),
|
||
("U4", [3]),
|
||
("U7", [4]),
|
||
("U11", [5]),
|
||
("U5", [6]),
|
||
("U6", [7]),
|
||
]
|
||
today = {
|
||
n: ("passed" if int(t) in (1, 3, 7) else "failed")
|
||
for t, spec in _CONFIG["feedback_types"].items()
|
||
for n in spec["evidence"]
|
||
}
|
||
row = gate.score_maf(maf, gate.green_types(_CONFIG["feedback_types"], today), gate._PACKAGE_SRC)
|
||
assert (row.k, row.n, row.status) == (3, 8, gate.RED)
|
||
assert all(p not in " ".join(row.exceptions) for p in ("U12 ", "U4 ", "U6 "))
|
||
|
||
|
||
def _synthetic_src(tmp: Path, *, comment_only: bool = False) -> Path:
|
||
body = (
|
||
"# uses SkillsProvider\n" if comment_only else "def build():\n return SkillsProvider()\n"
|
||
)
|
||
_write(tmp / "skills.py", "from agent_framework import SkillsProvider\n\n" + body)
|
||
return tmp
|
||
|
||
|
||
def _one_point(types: list[int]) -> dict[str, Any]:
|
||
point = {
|
||
"u_id": "U5",
|
||
"construct": "SkillsProvider",
|
||
"package": "agent_framework",
|
||
"callsite": {"module": "skills.py", "scope": "build"},
|
||
"types": types,
|
||
}
|
||
return {"approved": True, "points": [point]}
|
||
|
||
|
||
def test_row5_an_approved_point_counts_only_with_a_green_type(tmp_path: Path) -> None:
|
||
src = _synthetic_src(tmp_path)
|
||
green = gate.score_maf(_one_point([6]), {6: ""}, src)
|
||
assert (green.k, green.status) == (1, gate.GREEN)
|
||
red = gate.score_maf(_one_point([6]), {6: "failed"}, src)
|
||
assert (red.k, red.status) == (0, gate.RED)
|
||
|
||
|
||
def test_row5_a_comment_is_not_a_call_site(tmp_path: Path) -> None:
|
||
src = _synthetic_src(tmp_path, comment_only=True)
|
||
row = gate.score_maf(_one_point([6]), {6: ""}, src)
|
||
assert row.k == 0
|
||
assert "presence 0 av 1" in row.diagnostics
|
||
|
||
|
||
def test_row5_real_call_sites_are_found_by_ast() -> None:
|
||
found = {
|
||
p["u_id"]: gate.maf_presence(p, gate._PACKAGE_SRC) for p in _CONFIG["maf_points"]["points"]
|
||
}
|
||
assert found.pop("U5") == (False, False)
|
||
assert set(found.values()) == {(True, True)}
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Rows 6 and 7 — the stress artefacts
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
_PROBES = list(_CONFIG["row6_evidence"])
|
||
|
||
|
||
def test_row6_green_needs_both_the_probes_and_zero_undeclared() -> None:
|
||
row = gate.score_undeclared(_PROBES, _all_pass(_PROBES), _CLEAN, "s")
|
||
assert (row.k, row.n, row.status) == (0, 10, gate.GREEN)
|
||
dirty = gate.StressMeasure(validated=10, undeclared=3, undeclared_ids=("a4",), where="x")
|
||
assert gate.score_undeclared(_PROBES, _all_pass(_PROBES), dirty, "s").status == gate.RED
|
||
failing = {**_all_pass(_PROBES), _PROBES[1]: "failed"}
|
||
assert gate.score_undeclared(_PROBES, failing, _CLEAN, "s").status == gate.RED
|
||
|
||
|
||
def test_row6_missing_artefacts_are_never_zero() -> None:
|
||
row = gate.score_undeclared(
|
||
_PROBES, _all_pass(_PROBES), gate.StressMeasure(missing="ut finnes ikke"), "s"
|
||
)
|
||
assert row.k is None
|
||
assert "ikke målt, artefakter mangler" in row.reason
|
||
assert "– av –" in row.line()
|
||
|
||
|
||
def test_row6_measures_the_stress_outboxes_when_they_exist(tmp_path: Path) -> None:
|
||
"""Against the real artefacts when this machine has them; otherwise the absence is named."""
|
||
evidence = _CONFIG["stress_evidence"]
|
||
root = _REPO / evidence["root"]
|
||
bundles = Path("~/repos/vegnormal-okf/build/ferdig").expanduser()
|
||
if not root.is_dir() or not bundles.is_dir():
|
||
m = gate.measure_stress(evidence, _REPO, tmp_path / "absent", bundles)
|
||
assert m.missing and m.validated == 0
|
||
pytest.skip(f"stress artefacts not mounted ({root}, {bundles})")
|
||
m = gate.measure_stress(evidence, _REPO, root, bundles)
|
||
if m.missing:
|
||
# The mount belongs to another repository and can be mid-rebuild; the gate then says
|
||
# "ikke målt", which test_row6_missing_artefacts_are_never_zero already pins.
|
||
assert m.validated == 0
|
||
pytest.skip(f"stress artefacts not judgeable right now: {m.missing}")
|
||
assert (m.validated, m.undeclared, m.named, m.commissioned) == (10, 10, 1, 20)
|
||
|
||
|
||
def test_row7_is_a_diagnosis_and_never_moves_the_exit_code() -> None:
|
||
row = gate.score_named(_CLEAN, "s")
|
||
assert (row.k, row.n, row.status, row.failing) == (1, 20, gate.DIAGNOSIS, False)
|
||
assert gate.NAMED_WARNING in row.diagnostics
|
||
assert gate.exit_code([row]) == 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# The whole gate
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
def test_every_failing_row_green_is_exit_zero_and_one_red_is_exit_one(tmp_path: Path) -> None:
|
||
config = json.loads(json.dumps(_CONFIG))
|
||
config["maf_points"]["approved"] = True
|
||
config["maf_points"]["points"] = _one_point([6])["points"]
|
||
rows = gate.evaluate(
|
||
rounds_dir=_green_rounds(tmp_path),
|
||
config=config,
|
||
repo_root=_REPO,
|
||
src=_synthetic_src(tmp_path / "src"),
|
||
probe_runner=_all_pass,
|
||
stress_measure=_CLEAN,
|
||
)
|
||
assert [r.status for r in rows[:6]] == [gate.GREEN] * 6, gate.render(rows)
|
||
assert gate.exit_code(rows) == 0
|
||
(tmp_path / "3" / "report.kept.md").unlink()
|
||
rows = gate.evaluate(
|
||
rounds_dir=tmp_path,
|
||
config=config,
|
||
repo_root=_REPO,
|
||
src=tmp_path / "src",
|
||
probe_runner=_all_pass,
|
||
stress_measure=_CLEAN,
|
||
)
|
||
assert gate.exit_code(rows) == 1
|
||
|
||
|
||
def test_the_default_rounds_dir_is_gitignored() -> None:
|
||
proc = subprocess.run(
|
||
["git", "check-ignore", "-q", gate.DEFAULT_ROUNDS_DIR + "/1/feedback.json"], cwd=_REPO
|
||
)
|
||
assert proc.returncode == 0
|
||
|
||
|
||
def _cli(*args: str) -> subprocess.CompletedProcess[str]:
|
||
return subprocess.run(
|
||
[sys.executable, "-m", "portfolio_optimiser.evals.v1_gate", *args],
|
||
cwd=_REPO,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
|
||
|
||
def test_wrong_usage_is_exit_two(tmp_path: Path) -> None:
|
||
assert _cli("--rounds-dir", str(tmp_path / "missing")).returncode == 2
|
||
assert _cli("--no-such-flag").returncode == 2
|
||
help_text = _cli("--help").stdout
|
||
assert "report.kept.md" in help_text and "feedback_ids" in help_text
|
||
|
||
|
||
def test_the_command_is_red_today_with_every_row_in_its_output(tmp_path: Path) -> None:
|
||
proc = _cli("--rounds-dir", str(tmp_path), "--json")
|
||
assert proc.returncode == 1, proc.stderr
|
||
payload = json.loads(proc.stdout)
|
||
assert payload["exit"] == 1
|
||
rows = {r["key"]: r for r in payload["rows"]}
|
||
assert list(rows) == ["rounds", "changes", "types", "kept", "maf", "undeclared", "named"]
|
||
assert (rows["rounds"]["k"], rows["changes"]["k"]) == (0, 0)
|
||
assert (rows["types"]["k"], rows["types"]["n"]) == (3, 8)
|
||
assert rows["kept"]["status"] == gate.RED
|
||
assert (rows["maf"]["k"], rows["maf"]["n"], rows["maf"]["status"]) == (3, 8, gate.RED)
|
||
assert rows["undeclared"]["status"] == gate.RED
|
||
assert rows["named"]["failing"] is False
|