Kjell Tore Guttormsen
ab00016ed5
M-6 moved the price of a forgery from `touch` to reproducing the product's own artefact
set. The PM checkpoint of 18.09 then paid that price three ways, and row 2 read 3 of 3
GREEN each time:
F1 the whole artefact family written by hand and made internally consistent — coverage,
one proposal/outcome pair per approach, `verdict_id` minted with the product's own
rule. About sixty lines of script. `RUN_ATTESTATION` stood in the output; the row was
green anyway.
F2 four REAL runs' artefacts under a handwritten feedback file.
F4 `<n>/outbox` made a symlink to a real run's directory somewhere else, so a DERIVED
path was redirected by the filesystem.
F4 is a hole. F1 and F2 are not: no arrangement of files can be told apart from a round
that happened, because what is missing is not a check but a WITNESS. These tests say so.
Red here, 16 of them:
- rows 1 and 2 must stop at `gate.FORM_OK` — 0 against the criterion, exit still 1 — on a
tree where every computed check passes but nobody has attested the rounds (F1, F2);
- F2 is built through `outbox.write_outbox` and `write_coverage`, the exact bytes a real
run leaves behind, so it doubles as the control that `verify_run` reads the PRODUCT's
output and not merely the shape this test file writes;
- an outbox that is a symlink out of the round, and an artefact symlinked into one, are
refused by name (F4);
- the step from FORM OK to GREEN is `<n>/attestering.txt`, per round, round 0 included
because row 2 measures round 1 against it: present and matching -> GREEN (the rc-0
control, asserted first in every arm), missing -> FORM OK, present but naming another
round, another run, no date, an unparsable date or a date before the run -> RED;
- nothing in `src/` may write that file: a product that can produce a witness to its own
run has produced exactly the thing rows 1-2 cannot.
One more arm, green on arrival, and that is the finding: the guard at `v1_gate.py:415`
(`keyed != {(run_id, aid)}`), pinned to a constant-false branch, left the suite at 68
passed, identical to baseline. The guard bit, nothing read it — so
`_mut_the_family_is_labelled_for_another_run` gives the pair the right FILENAME and
another run's labels inside.
The fixture now writes the operator's attestation on every round it builds, and the
file's name is pinned in the test file as well (`_ATTEST_FILE`), the way M-5 pins the
rest of the contract: renaming it in the gate alone must fail a test.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1267 lines
53 KiB
Python
1267 lines
53 KiB
Python
"""The v1 gate's own tests: every row CAN go green and CAN go red — and cannot be FAKED green.
|
||
|
||
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. An independent review (17.09) then showed three rows
|
||
could be made green from a handwritten directory in a minute, and ten of twenty mutants survived
|
||
this file; the arms marked M-1 … M-5 and m-1 are the answer, each named for the finding it pins.
|
||
|
||
The probes and the stress measurement are injected here (``probe_runner`` / ``stress_measure``) so
|
||
the logic is exercised without a child pytest; ``run_probes`` gets its own arm against a throwaway
|
||
test file, and one subprocess arm runs the real command end to end.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from portfolio_optimiser import frozen_bundles
|
||
from portfolio_optimiser.evals import v1_gate as gate
|
||
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
|
||
from portfolio_optimiser.outbox import write_coverage, write_outbox
|
||
from portfolio_optimiser.provenance import Citation, ProvenanceStamp
|
||
from portfolio_optimiser.retrieval import TextSpan
|
||
from portfolio_optimiser.validator import UNSUPPORTED_REASON, Rejection, ValidatedProposal
|
||
from portfolio_optimiser.verdicts import features_from_ir, verdict_key
|
||
|
||
_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"]
|
||
)
|
||
_T0 = 1_780_000_000 # a fixed epoch: every fixture run and feedback is ordered against it
|
||
#: The day the operator attests a fixture round on: the fixture runs are all 2026-05-28.
|
||
_ATTEST_DATE = "2026-05-29"
|
||
#: The operator's attestation file, pinned HERE as well: it is the one name in the contract a
|
||
#: person types by hand, so renaming it in the gate alone must show up as a failing test.
|
||
_ATTEST_FILE = "attestering.txt"
|
||
|
||
#: The reason text a rejection at each stage carries, so a fixture run's coverage produces the
|
||
#: stage ``validator.rejection_stage`` would read off a real run.
|
||
_DETAIL = {
|
||
"stage0-baseline": "unknown cost code 'X': not in project P's cost baseline (1 known codes)",
|
||
"stage4-p90": "claimed 9 exceeds P90 feasible 1",
|
||
"unsupported": UNSUPPORTED_REASON,
|
||
}
|
||
|
||
|
||
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 _iso(offset: int) -> str:
|
||
from datetime import datetime, timezone
|
||
|
||
return datetime.fromtimestamp(_T0 + offset, tz=timezone.utc).isoformat()
|
||
|
||
|
||
def _feedback(
|
||
round_dir: Path,
|
||
*items: tuple[str, int, str],
|
||
author: str = "fagperson",
|
||
at: int | None = None,
|
||
**extra: Any,
|
||
) -> None:
|
||
offset = (int(round_dir.name) * 20 - 10) if at is None else at
|
||
_write(
|
||
round_dir / "feedback.json",
|
||
{
|
||
"author": author,
|
||
"given_at": _iso(offset),
|
||
"items": [{"id": i, "type": t, "text": x} for i, t, x in items],
|
||
**extra,
|
||
},
|
||
)
|
||
|
||
|
||
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 _coverage_row(row: dict[str, Any]) -> dict[str, Any]:
|
||
if row["validated"]:
|
||
return {
|
||
"id": row["id"],
|
||
"status": "validated",
|
||
"detail": "",
|
||
"saving_nok": row["validated_nok"],
|
||
}
|
||
if row["stage"] == "not_evaluated":
|
||
return {"id": row["id"], "status": "not_evaluated", "detail": "budget", "saving_nok": None}
|
||
status = "unsupported" if row["stage"] == "unsupported" else "rejected"
|
||
return {"id": row["id"], "status": status, "detail": _DETAIL[row["stage"]], "saving_nok": None}
|
||
|
||
|
||
def _ir(aid: str, nok: float | None) -> dict[str, Any]:
|
||
"""The proposal IR a run would have written for this approach — built through the real model
|
||
so the fixture cannot drift from the shape ``outbox.write_outbox`` actually persists, which is
|
||
the shape the gate re-mints the verdict key from."""
|
||
saving = 1.0 if nok is None else float(nok)
|
||
return SavingsProposal(
|
||
project_id="p",
|
||
measure=f"tiltak {aid}",
|
||
affected_items=[AffectedItem(code=f"K-{aid}", quantity=1.0, unit_cost=saving)],
|
||
claimed_saving_nok=saving,
|
||
).model_dump()
|
||
|
||
|
||
def _run_family(outbox: Path, run_id: str, coverage: list[dict[str, Any]]) -> None:
|
||
"""One proposal/outcome pair per EVALUATED approach, agreeing with its coverage row — the
|
||
artefact family ``gate.verify_run`` re-checks. ``verdict_id`` is minted with the product's own
|
||
rule (A5), because that is the field the gate mints again rather than reads."""
|
||
for cov in coverage:
|
||
if cov["status"] == "not_evaluated":
|
||
continue
|
||
aid = cov["id"]
|
||
ir = _ir(aid, cov.get("saving_nok"))
|
||
stem = f"{run_id}-{aid}"
|
||
_write(
|
||
outbox / f"{stem}-proposal.json",
|
||
{
|
||
"run_id": run_id,
|
||
"approach_id": aid,
|
||
"proposal": ir,
|
||
"provenance": {
|
||
"validator_decision": (
|
||
"rejected" if cov["status"] == "rejected" else "validated"
|
||
)
|
||
},
|
||
},
|
||
)
|
||
_write(
|
||
outbox / f"{stem}-outcome.json",
|
||
{
|
||
"run_id": run_id,
|
||
"approach_id": aid,
|
||
"outcome_type": cov["status"],
|
||
"reason": cov.get("detail", ""),
|
||
"verdict_id": verdict_key(features_from_ir(ir)),
|
||
},
|
||
)
|
||
|
||
|
||
def _outcome(
|
||
round_dir: Path,
|
||
rows: list[dict[str, Any]],
|
||
removed: Any = (),
|
||
*,
|
||
at: int | None = None,
|
||
coverage: list[dict[str, Any]] | None = None,
|
||
) -> None:
|
||
"""The round's outcome file AND the run it names: the run's OWN outbox inside the round,
|
||
holding its coverage and one artefact pair per evaluated approach. The run's time is DECLARED
|
||
(``ran_at``) — an mtime is not evidence, and the gate no longer reads one."""
|
||
run_id = f"r{round_dir.name}"
|
||
outbox = round_dir / "outbox"
|
||
if outbox.is_dir():
|
||
shutil.rmtree(outbox) # a re-written round brings its own run, not the last one's leftovers
|
||
rendered = [_coverage_row(r) for r in rows] if coverage is None else coverage
|
||
_write(
|
||
outbox / f"{run_id}-coverage.json",
|
||
{"run_id": run_id, "rows": rendered, "stop_reason": ""},
|
||
)
|
||
_run_family(outbox, run_id, rendered)
|
||
_write(
|
||
round_dir / "outcome.json",
|
||
{
|
||
"run_id": run_id,
|
||
"ran_at": _iso(int(round_dir.name) * 20 if at is None else at),
|
||
"approaches": rows,
|
||
"removed": list(removed),
|
||
},
|
||
)
|
||
|
||
|
||
def _attest(round_dir: Path, *, run_id: str | None = None, on: str = _ATTEST_DATE) -> None:
|
||
"""The operator's attestation for one round — the one file in the contract no machine may be
|
||
able to produce. Written here only because a fixture has to stand in for the operator; the
|
||
gate never writes one, and neither does anything else in the product."""
|
||
_write(
|
||
round_dir / _ATTEST_FILE,
|
||
f"runde: {round_dir.name}\n"
|
||
f"kjøring: {run_id or 'r' + round_dir.name}\n"
|
||
f"dato: {on}\n"
|
||
"Jeg bekrefter at denne runden ble holdt slik artefaktene beskriver.\n",
|
||
)
|
||
|
||
|
||
def _green_rounds(root: Path) -> Path:
|
||
"""Three traced rounds, each run after its feedback, a round 3 report the expert kept whole
|
||
while adding a line of their own — and the operator's attestation on every round, including
|
||
the baseline round 0 that round 1 is measured against."""
|
||
_outcome(root / "0", [_row("a1", False, None, stage="stage0-baseline")])
|
||
_write(root / "0" / "report.md", "# Rapport 0\n\nlinje\n")
|
||
_attest(root / "0")
|
||
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")
|
||
_attest(root / str(n))
|
||
_write(root / "3" / "report.kept.md", "# Rapport 3\n\nlinje\nmin egen merknad\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"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# M-5 — the contract is pinned to its SOURCE, not to whatever the data file says today
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
#: Operator's choice of the v1 destination, 16.09: three feedback rounds with a real expert.
|
||
_ROUNDS_REQUIRED = 3
|
||
#: Operator's choice, 16.09: the expert keeps at least 80 % of the round 3 report.
|
||
_KEEP_THRESHOLD = 0.8
|
||
#: The eight feedback types of the v1 outcome basis (item 9 is a note, not a type).
|
||
_TYPES = {"1", "2", "3", "4", "5", "6", "7", "8"}
|
||
#: Operator-approved 17.09: M = 8 U-IDs and the types each points at.
|
||
_M_LIST = [
|
||
("U13", [1, 2]),
|
||
("U9", [1, 8]),
|
||
("U12", [1]),
|
||
("U4", [3]),
|
||
("U7", [4]),
|
||
("U11", [5]),
|
||
("U5", [6]),
|
||
("U6", [7]),
|
||
]
|
||
#: The evidence each type is judged on. Changing it moves a type's verdict, so it is a contract
|
||
#: change: rewriting a probe goes through the PM checkpoint, and this list is where that shows.
|
||
_EVIDENCE = {
|
||
"1": [
|
||
"tests/test_proposal_review_loop_loadbearing.py::"
|
||
"test_t13_the_flag_answers_the_review_from_a_real_argv_and_the_answer_is_used"
|
||
],
|
||
"2": ["tests/test_v1_probes.py::test_type_2_remove_a_direction_has_a_typed_door"],
|
||
"3": [
|
||
"tests/test_v1_probes.py::test_type_3_a_commissioned_angle_is_evaluated_through_the_cli",
|
||
"tests/test_v1_probes.py::test_type_3_a_new_angle_changes_the_outcome",
|
||
],
|
||
"4": ["tests/test_v1_probes.py::test_type_4_relax_a_requirement_has_a_door"],
|
||
"5": ["tests/test_v1_probes.py::test_type_5_edit_the_concept_graph_has_a_door"],
|
||
"6": ["tests/test_v1_probes.py::test_type_6_skills_per_analysis_has_a_door"],
|
||
"7": [
|
||
"tests/test_v1_probes.py::"
|
||
"test_type_7_the_mcp_flag_puts_a_service_the_run_calls_into_the_result",
|
||
"tests/test_b4_mcp_call_trace_loadbearing.py::test_a_called_mcp_tool_is_recorded_in_provenance",
|
||
],
|
||
"8": ["tests/test_v1_probes.py::test_type_8_inline_context_has_a_door"],
|
||
}
|
||
|
||
|
||
def test_m5_the_contract_numbers_match_their_source() -> None:
|
||
assert _CONFIG["rounds_required"] == _ROUNDS_REQUIRED
|
||
assert _CONFIG["keep_threshold"] == _KEEP_THRESHOLD
|
||
assert set(_CONFIG["feedback_types"]) == _TYPES
|
||
assert _CONFIG["ai_authored"] == ["docs/ekspert-svar.md"]
|
||
maf = _CONFIG["maf_points"]
|
||
assert [(p["u_id"], p["types"]) for p in maf["points"]] == _M_LIST
|
||
assert (maf["approved"], maf["approved_on"], maf["approved_by"]) == (
|
||
True,
|
||
"2026-09-17",
|
||
"operatørgodkjent",
|
||
)
|
||
|
||
|
||
def test_m5_the_evidence_register_is_pinned() -> None:
|
||
assert {k: v["evidence"] for k, v in _CONFIG["feedback_types"].items()} == _EVIDENCE
|
||
assert _CONFIG["row6_evidence"] == [
|
||
"tests/test_v1_probes.py::test_row6_an_approach_that_declared_nothing_cannot_be_validated",
|
||
"tests/test_v1_probes.py::test_row6_a_run_level_declaration_does_not_stand_in_for_the_approach",
|
||
]
|
||
|
||
|
||
def test_m5_evaluate_uses_the_configured_numbers(tmp_path: Path) -> None:
|
||
"""The pins above would be decoration if ``evaluate`` hard-coded its own numbers."""
|
||
config = json.loads(json.dumps(_CONFIG))
|
||
config["rounds_required"] = 1
|
||
rows = gate.evaluate(
|
||
rounds_dir=tmp_path,
|
||
config=config,
|
||
repo_root=_REPO,
|
||
probe_runner=_all_pass,
|
||
stress_measure=_CLEAN,
|
||
)
|
||
assert (rows[0].n, rows[1].n) == (1, 1)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# 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_one_round_is_not_three(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
(root / "2" / "feedback.json").unlink()
|
||
(root / "3" / "feedback.json").unlink()
|
||
row = gate.score_rounds(root, 3, _AI)
|
||
assert (row.k, row.status) == (1, gate.RED)
|
||
|
||
|
||
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": "x", "given_at": _iso(10), "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)
|
||
assert any("navngir ingen fagperson" 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 — also with its
|
||
case changed (M-1: one letter used to get it through) — and the expert's own words count."""
|
||
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)
|
||
lifted = lifted.lstrip("> ")
|
||
root = _green_rounds(tmp_path)
|
||
_feedback(root / "1", ("f1", 1, lifted))
|
||
_feedback(root / "2", ("f2", 1, "Se her: " + lifted.swapcase() + " 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
|
||
|
||
|
||
def test_m1_the_same_feedback_copied_into_every_round_counts_once(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
for n in (1, 2, 3):
|
||
_feedback(root / str(n), (f"f{n}", 1, "ok"))
|
||
row = gate.score_rounds(root, 3, _AI)
|
||
assert row.k == 1
|
||
assert sum("ingen punkt som ikke alt er gitt" in x for x in row.exceptions) == 2
|
||
|
||
|
||
def test_m1_ids_are_per_round(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_feedback(root / "2", ("f1", 1, "Et helt nytt punkt i runde to."))
|
||
assert "gjenbrukt" in " ".join(gate.score_rounds(root, 3, _AI).exceptions)
|
||
|
||
|
||
def test_m1_feedback_is_timestamped_and_ordered(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_write(
|
||
root / "1" / "feedback.json",
|
||
{
|
||
"author": "x",
|
||
"given_at": "2026-09-17T10:00:00",
|
||
"items": [{"id": "f1", "type": 1, "text": "t"}],
|
||
},
|
||
)
|
||
_feedback(root / "3", ("f3", 1, "Et nytt punkt, men gitt for tidlig."), at=5)
|
||
row = gate.score_rounds(root, 3, _AI)
|
||
assert row.k == 1
|
||
assert any("tidssone" in x for x in row.exceptions)
|
||
assert any("ikke etter forrige" in x for x in row.exceptions)
|
||
|
||
|
||
def test_m1_feedback_is_given_on_a_report(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
(root / "0" / "report.md").unlink()
|
||
row = gate.score_rounds(root, 3, _AI)
|
||
assert row.k == 2
|
||
assert any("0/report.md" in x for x in row.exceptions)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# 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), row.exceptions
|
||
assert "runde 0 =" in row.reason and "(kjøring 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, 5000.0)]) # changed nok, no feedback id
|
||
_outcome(root / "3", [_row("a1", True, 9000.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_and_is_measured_against_the_previous_round(
|
||
tmp_path: Path,
|
||
) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_outcome(root / "2", [_row("a1", True, 1000.0, "f2")]) # identical to round 1, not to round 0
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert row.k == 2
|
||
assert any("ingen endring" in x for x in row.exceptions)
|
||
|
||
|
||
def test_m1_a_change_below_one_percent_is_noise(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_outcome(root / "2", [_row("a1", True, 1009.99, "f2")])
|
||
assert gate.score_changes(root, 3, _AI).k == 2
|
||
_outcome(root / "2", [_row("a1", True, 1010.0, "f2")])
|
||
assert gate.score_changes(root, 3, _AI).k == 3
|
||
|
||
|
||
def test_m1_an_outcome_must_name_a_run_that_exists_and_agree_with_it(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
run = root / "1" / "outbox" / "r1-coverage.json"
|
||
run.unlink()
|
||
assert "finnes ikke" in " ".join(gate.score_changes(root, 3, _AI).exceptions)
|
||
_outcome(
|
||
root / "1",
|
||
[_row("a1", True, 1000.0, "f1")],
|
||
coverage=[{"id": "a1", "status": "rejected", "detail": _DETAIL["stage4-p90"]}],
|
||
)
|
||
assert "stemmer ikke" in " ".join(gate.score_changes(root, 3, _AI).exceptions)
|
||
|
||
|
||
def test_m1_an_empty_baseline_run_is_refused(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_outcome(root / "0", [], coverage=[])
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert "evaluerte ingen" in row.reason
|
||
assert row.k == 2
|
||
|
||
|
||
def test_m1_the_feedback_must_come_between_the_two_runs(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_outcome(root / "1", [_row("a1", True, 1000.0, "f1")], at=5) # ran before its feedback (10)
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert row.k < 3
|
||
assert any("ikke gitt mellom kjøring" in x for x in row.exceptions)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# M-6 — the 18.09 re-measurement: row 2 went green on four handwritten files
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
def _forged_rounds(root: Path) -> Path:
|
||
"""The 18.09 re-measurement's attack, rebuilt HERE rather than through ``_outcome`` so it
|
||
cannot drift with the fixtures: four handwritten ``outcome.json``, four handwritten
|
||
``<run_id>-coverage.json`` in an outbox the forger named in those same files, and ``os.utime``
|
||
for the ordering. No run of this product ever touched this tree."""
|
||
chosen = root / "min-utboks"
|
||
for n in (0, 1, 2, 3):
|
||
run_id = f"r{n}"
|
||
validated = n > 0
|
||
nok = 1000.0 * n if validated else None
|
||
_write(
|
||
chosen / f"{run_id}-coverage.json",
|
||
{
|
||
"run_id": run_id,
|
||
"stop_reason": "",
|
||
"rows": [
|
||
{
|
||
"id": "a1",
|
||
"status": "validated" if validated else "rejected",
|
||
"detail": "" if validated else _DETAIL["stage0-baseline"],
|
||
"saving_nok": nok,
|
||
}
|
||
],
|
||
},
|
||
)
|
||
stamp = _T0 + n * 20
|
||
os.utime(chosen / f"{run_id}-coverage.json", (stamp, stamp))
|
||
_write(
|
||
root / str(n) / "outcome.json",
|
||
{
|
||
"run_id": run_id,
|
||
"outbox": str(chosen),
|
||
"approaches": [
|
||
_row(
|
||
"a1",
|
||
validated,
|
||
nok,
|
||
*([f"f{n}"] if validated else []),
|
||
stage="" if validated else "stage0-baseline",
|
||
)
|
||
],
|
||
"removed": [],
|
||
},
|
||
)
|
||
_write(root / str(n) / "report.md", f"# Rapport {n}\n\nlinje\n")
|
||
if n:
|
||
_feedback(
|
||
root / str(n), (f"f{n}", 1, f"Tallet for linje {n} er feil, bruk kontrakten.")
|
||
)
|
||
return root
|
||
|
||
|
||
def test_m6_a_handwritten_outbox_is_not_a_run(tmp_path: Path) -> None:
|
||
"""Row 2 must not be satisfiable by files a forger wrote. Row 1 stays GREEN on the same tree —
|
||
the feedback there IS well formed, and that is what makes this an attack on row 2 rather than
|
||
a broken fixture. The control that the row can still go green is ``_green_rounds``, which
|
||
carries a whole run family (``test_row2_a_traced_change_counts``)."""
|
||
root = _forged_rounds(tmp_path)
|
||
assert gate.score_rounds(root, 3, _AI).status == gate.GREEN
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert (row.k, row.status) == (0, gate.RED), row.exceptions
|
||
|
||
|
||
def _edit(path: Path, **changes: Any) -> None:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
data.update(changes)
|
||
_write(path, data)
|
||
|
||
|
||
def _mut_declares_its_own_outbox(root: Path) -> None:
|
||
_edit(root / "1" / "outcome.json", outbox=str(root / "1" / "outbox"))
|
||
|
||
|
||
def _mut_coverage_is_another_run(root: Path) -> None:
|
||
_edit(root / "1" / "outbox" / "r1-coverage.json", run_id="r9")
|
||
|
||
|
||
def _mut_the_approach_has_no_artefact(root: Path) -> None:
|
||
(root / "1" / "outbox" / "r1-a1-outcome.json").unlink()
|
||
|
||
|
||
def _mut_the_key_is_not_the_irs(root: Path) -> None:
|
||
_edit(
|
||
root / "1" / "outbox" / "r1-a1-outcome.json",
|
||
verdict_id=verdict_key(features_from_ir(_ir("a1", 424242.0))),
|
||
)
|
||
|
||
|
||
def _mut_the_stamp_contradicts_the_row(root: Path) -> None:
|
||
path = root / "1" / "outbox" / "r1-a1-proposal.json"
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
data["provenance"]["validator_decision"] = "rejected"
|
||
_write(path, data)
|
||
|
||
|
||
def _mut_the_outcome_type_contradicts_the_row(root: Path) -> None:
|
||
_edit(root / "1" / "outbox" / "r1-a1-outcome.json", outcome_type="rejected")
|
||
|
||
|
||
def _mut_the_figure_is_not_the_proposals(root: Path) -> None:
|
||
"""The IR says one figure and the coverage another — with the key RE-MINTED off the changed
|
||
IR, so this arm is felled by the figures and not by the key check standing in front of it."""
|
||
path = root / "1" / "outbox" / "r1-a1-proposal.json"
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
data["proposal"] = _ir("a1", 424242.0)
|
||
_write(path, data)
|
||
_edit(
|
||
root / "1" / "outbox" / "r1-a1-outcome.json",
|
||
verdict_id=verdict_key(features_from_ir(data["proposal"])),
|
||
)
|
||
|
||
|
||
def _mut_the_reason_is_not_the_rows(root: Path) -> None:
|
||
_edit(root / "0" / "outbox" / "r0-a1-outcome.json", reason="en annen grunn")
|
||
|
||
|
||
def _mut_an_artefact_belongs_to_another_run(root: Path) -> None:
|
||
_run_family(
|
||
root / "1" / "outbox",
|
||
"r1",
|
||
[{"id": "a9", "status": "validated", "detail": "", "saving_nok": 5.0}],
|
||
)
|
||
|
||
|
||
def _mut_an_unevaluated_approach_has_artefacts(root: Path) -> None:
|
||
_outcome(
|
||
root / "1",
|
||
[_row("a1", True, 1000.0, "f1"), _row("a2", False, None, stage="not_evaluated")],
|
||
)
|
||
_run_family(
|
||
root / "1" / "outbox",
|
||
"r1",
|
||
[{"id": "a2", "status": "rejected", "detail": "x", "saving_nok": None}],
|
||
)
|
||
|
||
|
||
def _mut_the_family_is_labelled_for_another_run(root: Path) -> None:
|
||
"""The right FILENAME, another run's labels INSIDE. The guard on ``keyed`` has stood since
|
||
M-6, but nothing read it: pinned to a constant-false branch it left the suite at 68 passed,
|
||
identical to baseline (PM checkpoint 18.09). A guard no test reads is a guard the next edit
|
||
deletes."""
|
||
for kind in ("proposal", "outcome"):
|
||
_edit(root / "1" / "outbox" / f"r1-a1-{kind}.json", run_id="r9", approach_id="a9")
|
||
|
||
|
||
def _mut_the_run_time_has_no_zone(root: Path) -> None:
|
||
_edit(root / "1" / "outcome.json", ran_at="2026-09-18T10:00:00")
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("mutate", "marker"),
|
||
[
|
||
(_mut_declares_its_own_outbox, "outbox kan ikke oppgis"),
|
||
(_mut_coverage_is_another_run, "coverage-fila er kjøring"),
|
||
(_mut_the_approach_has_no_artefact, "r1-a1-outcome.json mangler"),
|
||
(_mut_the_key_is_not_the_irs, "verdict_id er ikke nøkkelen"),
|
||
(_mut_the_stamp_contradicts_the_row, "validator_decision"),
|
||
(_mut_the_outcome_type_contradicts_the_row, "utfallet er"),
|
||
(_mut_the_figure_is_not_the_proposals, "forslaget selv sier"),
|
||
(_mut_the_reason_is_not_the_rows, "utfallets grunn"),
|
||
(_mut_an_artefact_belongs_to_another_run, "coverage ikke nevner"),
|
||
(_mut_an_unevaluated_approach_has_artefacts, "ble ikke evaluert, men kjøringen skrev"),
|
||
(_mut_the_family_is_labelled_for_another_run, "artefaktene er merket"),
|
||
(_mut_the_run_time_has_no_zone, "ran_at er ikke et ISO"),
|
||
],
|
||
)
|
||
def test_m6_the_run_family_must_stand_up_to_itself(
|
||
tmp_path: Path, mutate: Any, marker: str
|
||
) -> None:
|
||
"""One binding at a time, on a tree that is green until the mutation lands — and each arm is
|
||
read back by the reason it was refused for, so an arm cannot pass because some OTHER rule
|
||
happened to fire. The rc-0 control is the unmutated tree, asserted first."""
|
||
root = _green_rounds(tmp_path)
|
||
assert gate.score_changes(root, 3, _AI).status == gate.GREEN
|
||
mutate(root)
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert row.status == gate.RED
|
||
assert any(marker in x for x in row.exceptions), (marker, row.exceptions)
|
||
|
||
|
||
def test_m6_an_mtime_is_not_evidence_and_no_longer_decides_anything(tmp_path: Path) -> None:
|
||
"""The forger's third move was ``os.utime``. Every artefact in a green tree is stamped far in
|
||
the future, in the wrong order, and row 2 does not move: the run's time is the round's
|
||
declared ``ran_at``, and the row says so out loud."""
|
||
root = _green_rounds(tmp_path)
|
||
for n, path in enumerate(sorted(root.rglob("*.json"))):
|
||
os.utime(path, (_T0 - n * 1000, _T0 - n * 1000))
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert (row.k, row.status) == (3, gate.GREEN), row.exceptions
|
||
assert row.attests == (gate.RUN_ATTESTATION, gate.ATTEST_RULE)
|
||
assert "bekrefter operatøren" in gate.RUN_ATTESTATION
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# M-7 — the 18.09 PM check: three CONSISTENT forgeries still read 3 of 3 GREEN
|
||
# ---------------------------------------------------------------------------------------------
|
||
#
|
||
# M-6 moved the price of a forgery from ``touch`` to reproducing the product's own artefact set.
|
||
# The PM then paid that price three ways and row 2 read 3 of 3 GREEN each time: F1 a whole family
|
||
# written by hand and made internally consistent, verdict key minted with the product's own rule
|
||
# (about sixty lines of script) · F2 four REAL runs' artefacts under a handwritten feedback file ·
|
||
# F4 ``<n>/outbox`` symlinked out of the round to a real run somewhere else.
|
||
#
|
||
# F4 is a hole and is closed below. F1 and F2 are not: no arrangement of files can be told from a
|
||
# round that happened, because what is missing is not a check but a WITNESS. So the computation
|
||
# stops at FORM OK — 0 against the criterion, exit still 1 — and the step to GREEN is a statement
|
||
# the operator makes, per round, in a file the gate never writes.
|
||
|
||
|
||
def _unattested(root: Path) -> Path:
|
||
for path in sorted(root.rglob(_ATTEST_FILE)):
|
||
path.unlink()
|
||
return root
|
||
|
||
|
||
def test_m7_a_consistent_family_is_form_ok_and_counts_zero(tmp_path: Path) -> None:
|
||
"""F1. Everything ``verify_run`` recomputes agrees — because the forger recomputed it too.
|
||
Rows 1 and 2 must read FORM OK, count 0 of 3, and leave the gate's exit at 1."""
|
||
root = _unattested(_green_rounds(tmp_path))
|
||
rounds, changes = gate.score_rounds(root, 3, _AI), gate.score_changes(root, 3, _AI)
|
||
assert (rounds.k, rounds.status) == (0, gate.FORM_OK), rounds.exceptions
|
||
assert (changes.k, changes.status) == (0, gate.FORM_OK), changes.exceptions
|
||
assert gate.exit_code([rounds, changes]) == 1
|
||
assert any(_ATTEST_FILE in x for x in rounds.exceptions), rounds.exceptions
|
||
|
||
|
||
def _stamp(decision: str) -> ProvenanceStamp:
|
||
return ProvenanceStamp(
|
||
citations=[
|
||
Citation(file="f.md", locator=TextSpan(start_index=0, end_index=5), snippet="h")
|
||
],
|
||
model="synthetic",
|
||
role="proposer",
|
||
validator_decision=decision, # type: ignore[arg-type]
|
||
token_usage=8,
|
||
cost_baseline_anchored=True,
|
||
bundle_id_source=None,
|
||
code_forms={},
|
||
)
|
||
|
||
|
||
def _real_run_family(round_dir: Path) -> None:
|
||
"""F2. The round's outbox rewritten BY THE PRODUCT — ``outbox.write_outbox`` and
|
||
``write_coverage``, the exact bytes a real run leaves behind — for the coverage the round
|
||
already declares. This is also the control that ``verify_run`` reads the product's own
|
||
output and not merely the shape this test file happens to write: if the two ever part, the
|
||
gate is checking a fixture."""
|
||
run_id = f"r{round_dir.name}"
|
||
outbox = round_dir / "outbox"
|
||
coverage = json.loads((outbox / f"{run_id}-coverage.json").read_text(encoding="utf-8"))["rows"]
|
||
shutil.rmtree(outbox)
|
||
for cov in coverage:
|
||
if cov["status"] == "not_evaluated":
|
||
continue
|
||
ir = _ir(cov["id"], cov.get("saving_nok"))
|
||
proposal = SavingsProposal(**ir)
|
||
validated = cov["status"] == "validated"
|
||
outcome: Any = (
|
||
ValidatedProposal(proposal=proposal, p10=1.0, p50=2.0, p90=3.0, nominal_feasible=2.0)
|
||
if validated
|
||
else Rejection(proposal=proposal, reason=cov["detail"])
|
||
)
|
||
write_outbox(
|
||
str(outbox),
|
||
run_id,
|
||
outcome=outcome,
|
||
provenance=_stamp("validated" if validated else "rejected"),
|
||
checker_verdict="approve",
|
||
verdict_id=verdict_key(features_from_ir(ir)),
|
||
approach_id=cov["id"],
|
||
)
|
||
write_coverage(str(outbox), run_id, rows=coverage, stop_reason="")
|
||
|
||
|
||
def test_m7_the_products_own_artefacts_do_not_prove_a_round_either(tmp_path: Path) -> None:
|
||
"""F2. Real artefacts, handwritten feedback. The rc-0 control comes first: with the operator's
|
||
attestation in place the SAME product-written tree is GREEN, so this arm fails on the missing
|
||
witness and not on a fixture the gate cannot read."""
|
||
root = _green_rounds(tmp_path)
|
||
for n in (0, 1, 2, 3):
|
||
_real_run_family(root / str(n))
|
||
assert gate.score_changes(root, 3, _AI).status == gate.GREEN, gate.score_changes(
|
||
root, 3, _AI
|
||
).exceptions
|
||
row = gate.score_changes(_unattested(root), 3, _AI)
|
||
assert (row.k, row.status) == (0, gate.FORM_OK), row.exceptions
|
||
|
||
|
||
def test_m7_an_outbox_that_leaves_the_round_is_refused(tmp_path: Path) -> None:
|
||
"""F4. ``<n>/outbox`` made a symlink to a real run's directory elsewhere, and the round read
|
||
that run as its own. A derived path is only derived if the filesystem cannot redirect it."""
|
||
root = _green_rounds(tmp_path / "rounds")
|
||
elsewhere = tmp_path / "en-ekte-kjoering"
|
||
shutil.copytree(root / "1" / "outbox", elsewhere)
|
||
shutil.rmtree(root / "1" / "outbox")
|
||
(root / "1" / "outbox").symlink_to(elsewhere, target_is_directory=True)
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert (row.k, row.status) == (1, gate.RED), row.exceptions
|
||
assert any("lenke ut av runden" in x for x in row.exceptions), row.exceptions
|
||
|
||
|
||
def test_m7_an_artefact_symlinked_into_the_round_is_refused(tmp_path: Path) -> None:
|
||
"""The same move one level down: the outbox is the round's own directory, but a file in it
|
||
points at another run's artefact."""
|
||
root = _green_rounds(tmp_path / "rounds")
|
||
target = tmp_path / "r1-a1-outcome.json"
|
||
artefact = root / "1" / "outbox" / "r1-a1-outcome.json"
|
||
shutil.copy(artefact, target)
|
||
artefact.unlink()
|
||
artefact.symlink_to(target)
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert row.status == gate.RED
|
||
assert any("lenke" in x for x in row.exceptions), row.exceptions
|
||
|
||
|
||
def test_m7_the_attestation_is_the_step_from_form_ok_to_green(tmp_path: Path) -> None:
|
||
"""The control in both directions: the same tree is GREEN with the operator's attestation and
|
||
FORM OK without it. A gate that refuses everything is as useless as one that refuses nothing
|
||
(P21 C1), and a row that can only read FORM OK would be exactly that."""
|
||
root = _green_rounds(tmp_path)
|
||
assert gate.score_rounds(root, 3, _AI).status == gate.GREEN
|
||
assert gate.score_changes(root, 3, _AI).status == gate.GREEN
|
||
(root / "2" / _ATTEST_FILE).unlink()
|
||
rounds, changes = gate.score_rounds(root, 3, _AI), gate.score_changes(root, 3, _AI)
|
||
assert (rounds.k, rounds.status) == (2, gate.FORM_OK), rounds.exceptions
|
||
assert (changes.k, changes.status) == (2, gate.FORM_OK), changes.exceptions
|
||
assert any("mangler" in x for x in rounds.exceptions), rounds.exceptions
|
||
|
||
|
||
def test_m7_row2_needs_the_base_run_attested_too(tmp_path: Path) -> None:
|
||
"""Round 1 is measured against round 0's run, so an unattested baseline is an unattested
|
||
comparison and row 2 counts nothing. Row 1, which reads no run at all, is untouched."""
|
||
root = _green_rounds(tmp_path)
|
||
(root / "0" / _ATTEST_FILE).unlink()
|
||
assert gate.score_rounds(root, 3, _AI).status == gate.GREEN
|
||
row = gate.score_changes(root, 3, _AI)
|
||
assert (row.k, row.status) == (0, gate.FORM_OK), row.exceptions
|
||
assert any("grunnkjøringen" in x for x in row.exceptions), row.exceptions
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("text", "marker"),
|
||
[
|
||
("runde: 1\nkjøring: r9\ndato: 2026-05-29\n", "navngir kjøring"),
|
||
("runde: 3\nkjøring: r1\ndato: 2026-05-29\n", "attesterer runde"),
|
||
("runde: 1\nkjøring: r1\n", "mangler dato"),
|
||
("kjøring: r1\ndato: 2026-05-29\n", "mangler runde"),
|
||
("runde: 1\nkjøring: r1\ndato: i går\n", "er ikke en ISO-dato"),
|
||
("runde: 1\nkjøring: r1\ndato: 2020-01-01\n", "før kjøringen"),
|
||
("Jeg bekrefter at runde 1 ble holdt.\n", "mangler runde, kjøring, dato"),
|
||
],
|
||
)
|
||
def test_m7_an_attestation_that_contradicts_the_round_is_red(
|
||
tmp_path: Path, text: str, marker: str
|
||
) -> None:
|
||
"""A MISSING attestation is FORM OK — nobody has confirmed anything yet. A PRESENT one that
|
||
does not match the round is something else: a statement about a round this is not, and the
|
||
row goes red rather than waiting."""
|
||
root = _green_rounds(tmp_path)
|
||
assert gate.score_rounds(root, 3, _AI).status == gate.GREEN
|
||
_write(root / "1" / _ATTEST_FILE, text)
|
||
row = gate.score_rounds(root, 3, _AI)
|
||
assert (row.k, row.status) == (2, gate.RED), row.exceptions
|
||
assert any(marker in x for x in row.exceptions), (marker, row.exceptions)
|
||
|
||
|
||
def test_m7_the_attestation_file_is_pinned() -> None:
|
||
assert gate.ATTEST_FILE == _ATTEST_FILE
|
||
|
||
|
||
def test_m7_the_gate_never_writes_an_attestation(tmp_path: Path) -> None:
|
||
"""The file is the operator's word. A product that can produce one has produced a witness to
|
||
its own run, which is the whole thing rows 1-2 cannot do: so no source file in the package
|
||
writes ``ATTEST_FILE``, and scoring a tree leaves none behind."""
|
||
root = _unattested(_green_rounds(tmp_path))
|
||
gate.score_rounds(root, 3, _AI)
|
||
gate.score_changes(root, 3, _AI)
|
||
assert list(root.rglob(_ATTEST_FILE)) == []
|
||
writes = [
|
||
f"{path.name}:{i}"
|
||
for path in sorted((_REPO / "src" / "portfolio_optimiser").rglob("*.py"))
|
||
for i, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1)
|
||
if _ATTEST_FILE in line and "write_text" in line
|
||
]
|
||
assert writes == []
|
||
|
||
|
||
def _outcome_obj(rows: list[dict[str, Any]], removed: dict[str, set[str]] | None = None) -> Any:
|
||
from datetime import datetime, timezone
|
||
|
||
return gate.Outcome(
|
||
{r["id"]: r for r in rows}, removed or {}, "r", datetime.now(tz=timezone.utc)
|
||
)
|
||
|
||
|
||
def test_row2_each_of_a_to_d_is_a_change() -> None:
|
||
ids = {"f"}
|
||
base = [_row("a1", False, None, stage="stage0-baseline")]
|
||
cases = {
|
||
"a-added": [*base, _row("a2", False, None, "f", stage="stage0-baseline")],
|
||
"b-validated": [_row("a1", True, 7.0, "f")],
|
||
"c-stage": [_row("a1", False, None, "f", stage="stage4-p90")],
|
||
"d-nok": [_row("a1", False, 5.0, "f", stage="stage0-baseline")],
|
||
}
|
||
for name, rows in cases.items():
|
||
ok, why = gate.outcomes_changed(_outcome_obj(base), _outcome_obj(rows), ids)
|
||
assert ok, (name, why)
|
||
before = _outcome_obj([*base, _row("a2", False, None)])
|
||
assert not gate.outcomes_changed(before, _outcome_obj(base), ids)[0]
|
||
removed = _outcome_obj(base, {"a2": {"f"}})
|
||
assert gate.outcomes_changed(before, removed, 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
|
||
|
||
|
||
def test_m1_run_probes_reads_every_outcome_honestly(tmp_path: Path) -> None:
|
||
"""m-1 (M06-M08, M16): an erroring or skipped test is not a pass, and a known gap marked
|
||
``xfail(strict=True)`` is run as the failure it is."""
|
||
_write(
|
||
tmp_path / "tests" / "test_probe.py",
|
||
"import pytest\n\n"
|
||
"@pytest.fixture\n"
|
||
"def broken():\n raise RuntimeError('fixture')\n\n"
|
||
"def test_pass():\n pass\n\n"
|
||
"def test_fail():\n assert False\n\n"
|
||
"def test_error(broken):\n pass\n\n"
|
||
"def test_skip():\n pytest.skip('no')\n\n"
|
||
"@pytest.mark.xfail(strict=True)\n"
|
||
"def test_gap():\n assert False\n",
|
||
)
|
||
names = ["pass", "fail", "error", "skip", "gap", "gone"]
|
||
ids = [f"tests/test_probe.py::test_{n}" for n in names]
|
||
outcomes = gate.run_probes(ids, repo_root=tmp_path)
|
||
assert [outcomes[i] for i in ids] == [
|
||
"passed",
|
||
"failed",
|
||
"failed",
|
||
"skipped",
|
||
"failed",
|
||
"missing",
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Row 4 — round 3 kept
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
def _report(root: Path, report: str, kept: str | None) -> None:
|
||
_write(root / "3" / "report.md", report)
|
||
if kept is not None:
|
||
_write(root / "3" / "report.kept.md", kept)
|
||
|
||
|
||
def _kept_row(root: Path, report: str, kept: str) -> gate.Row:
|
||
_report(root, report, kept)
|
||
return gate.score_kept(root, 0.8, _AI)
|
||
|
||
|
||
@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)]
|
||
kept = lines[:kept_lines] + [f"endret {i}" for i in range(100 - kept_lines)]
|
||
_report(tmp_path, "\n\n".join(lines) + "\n", "\n".join(kept) + "\n")
|
||
row = gate.score_kept(tmp_path, 0.8, _AI)
|
||
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:
|
||
_report(tmp_path, "a\nb\n", None)
|
||
row = gate.score_kept(tmp_path, 0.8, _AI)
|
||
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:
|
||
_report(tmp_path, "x\nx\ny\n", "x\nz\n")
|
||
assert gate.score_kept(tmp_path, 0.8, _AI).k == 1
|
||
|
||
|
||
def test_m2_separators_do_not_make_a_rewritten_report_kept(tmp_path: Path) -> None:
|
||
rules = "---\n| --- | --- |\n \n" * 34
|
||
report = rules + "".join(f"innhold {i}\n" for i in range(5))
|
||
kept = rules + "".join(f"omskrevet {i}\n" for i in range(5))
|
||
row = _kept_row(tmp_path, report, kept)
|
||
assert (row.k, row.n, row.status) == (0, 5, gate.RED)
|
||
|
||
|
||
def test_m2_trailing_whitespace_is_an_editor_not_an_edit(tmp_path: Path) -> None:
|
||
report = "".join(f"linje {i} \n" for i in range(10))
|
||
row = _kept_row(tmp_path, report, report.replace(" \n", "\n"))
|
||
assert (row.k, row.n, row.status) == (10, 10, gate.GREEN)
|
||
|
||
|
||
def test_m2_a_reshuffle_is_a_change(tmp_path: Path) -> None:
|
||
lines = [f"linje {i}" for i in range(10)]
|
||
row = _kept_row(tmp_path, "\n".join(lines), "\n".join(reversed(lines)))
|
||
assert row.k == 1 and row.status == gate.RED
|
||
|
||
|
||
def test_m2_additions_are_their_own_number(tmp_path: Path) -> None:
|
||
report = "".join(f"linje {i}\n" for i in range(10))
|
||
kept = report + "".join(f"innvending {i}\n" for i in range(200))
|
||
row = _kept_row(tmp_path, report, kept)
|
||
assert (row.k, row.status) == (10, gate.GREEN)
|
||
assert "200 linje(r) lagt til" in row.reason
|
||
|
||
|
||
def test_m2_an_untouched_copy_needs_a_receipt(tmp_path: Path) -> None:
|
||
report = "".join(f"linje {i}\n" for i in range(10))
|
||
row = _kept_row(tmp_path, report, report)
|
||
assert (row.k, row.status) == (None, gate.RED)
|
||
assert "ikke rørt" in row.reason
|
||
_feedback(tmp_path / "3", ("f3", 1, "Rapporten kan stå som den er."), report_unchanged=True)
|
||
row = gate.score_kept(tmp_path, 0.8, _AI)
|
||
assert (row.k, row.status) == (10, gate.GREEN)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# 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:
|
||
maf = _CONFIG["maf_points"]
|
||
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, body: str | None = None) -> Path:
|
||
body = "def build():\n return SkillsProvider()\n" if body is None else body
|
||
_write(tmp / "skills.py", "from agent_framework import SkillsProvider\n\n" + body)
|
||
return tmp
|
||
|
||
|
||
def _one_point(types: list[int], scope: str = "build") -> dict[str, Any]:
|
||
point = {
|
||
"u_id": "U5",
|
||
"construct": "SkillsProvider",
|
||
"package": "agent_framework",
|
||
"callsite": {"module": "skills.py", "scope": scope},
|
||
"types": types,
|
||
}
|
||
return {"approved": True, "points": [point]}
|
||
|
||
|
||
def test_row5_an_approved_point_counts_only_when_every_type_it_points_at_is_green(
|
||
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)
|
||
assert gate.score_maf(_one_point([6]), {6: "failed"}, src).k == 0
|
||
assert gate.score_maf(_one_point([6, 7]), {6: "", 7: "failed"}, src).k == 0
|
||
|
||
|
||
def test_row5_a_comment_is_not_a_call_site(tmp_path: Path) -> None:
|
||
src = _synthetic_src(tmp_path, "# uses SkillsProvider\n")
|
||
row = gate.score_maf(_one_point([6]), {6: ""}, src)
|
||
assert row.k == 0
|
||
assert "presence 0 av 1" in row.diagnostics
|
||
|
||
|
||
def test_row5_the_named_scope_is_required(tmp_path: Path) -> None:
|
||
src = _synthetic_src(tmp_path, "def other():\n return SkillsProvider()\n")
|
||
row = gate.score_maf(_one_point([6]), {6: ""}, src)
|
||
assert row.k == 0
|
||
assert "kallsted verifisert 0 av 1" in row.diagnostics
|
||
|
||
|
||
def test_row5_a_type_annotation_is_not_a_use(tmp_path: Path) -> None:
|
||
src = _synthetic_src(
|
||
tmp_path, "def build(p: SkillsProvider) -> SkillsProvider:\n return p\n"
|
||
)
|
||
assert gate.maf_presence(_one_point([6])["points"][0], src) == (False, False)
|
||
|
||
|
||
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
|
||
assert gate.score_undeclared([], {}, _CLEAN, "s").status == gate.RED
|
||
|
||
|
||
def test_row6_missing_artefacts_are_never_zero_and_never_green() -> None:
|
||
"""Probes passing and artefacts absent (clean clone, CI, a base mid-rebuild) is NOT a pass:
|
||
the row says IKKE MÅLT and fails the exit code."""
|
||
row = gate.score_undeclared(
|
||
_PROBES, _all_pass(_PROBES), gate.StressMeasure(missing="ut finnes ikke"), "s"
|
||
)
|
||
assert row.k is None
|
||
assert "ikke målt" in row.reason
|
||
assert "– av –" in row.line()
|
||
assert (row.status, row.failing) == (gate.NOT_MEASURED, True)
|
||
assert gate.exit_code([row]) == 1
|
||
|
||
|
||
def test_row6_artefacts_older_than_the_rule_are_not_measured() -> None:
|
||
old = gate.StressMeasure(validated=10, undeclared=10, unaddressed=12, where="x")
|
||
row = gate.score_undeclared(_PROBES, _all_pass(_PROBES), old, "s")
|
||
assert (row.k, row.status) == (None, gate.NOT_MEASURED)
|
||
assert "eldre enn regelen" in row.reason and "approach_id mangler" in row.reason
|
||
|
||
|
||
def test_m3_own_proposals_are_in_the_denominator(tmp_path: Path) -> None:
|
||
runs = {"runs": [{"outbox": "o", "run_id": "r1"}, {"outbox": "o", "run_id": "r2"}]}
|
||
_write(tmp_path / "o" / "r1-own-proposal-outcome.json", {"outcome_type": "validated"})
|
||
_write(tmp_path / "o" / "r2-own-proposal-outcome.json", {"outcome_type": "validated"})
|
||
_write(
|
||
tmp_path / "o" / "r2-debate.json",
|
||
{"requirements": [{"path": "p", "approach_id": "own-proposal"}]},
|
||
)
|
||
assert gate._own_proposals(runs, tmp_path) == (2, 1, ("own-proposal (r1)",))
|
||
|
||
|
||
def test_row7_not_measured_is_not_green_either() -> None:
|
||
row = gate.score_named(gate.StressMeasure(missing="borte"), "s")
|
||
assert (row.k, row.status, row.failing) == (None, gate.NOT_MEASURED, False)
|
||
|
||
|
||
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 = frozen_bundles.store_root()
|
||
if not root.is_dir() or not bundles.is_dir():
|
||
m = gate.measure_stress(evidence, _REPO, tmp_path / "absent", None)
|
||
assert m.missing and m.validated == 0
|
||
pytest.skip(f"stress artefacts not mounted ({root}, {bundles})")
|
||
m = gate.measure_stress(evidence, _REPO, root, None)
|
||
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}")
|
||
# 10 commissioned + 5 of the runs' own proposals (M-3).
|
||
assert (m.validated, m.undeclared, m.own_validated, m.named, m.commissioned) == (
|
||
15,
|
||
15,
|
||
5,
|
||
1,
|
||
20,
|
||
)
|
||
assert m.unaddressed > 0 # stress round 6 predates approach-addressed declarations
|
||
row = gate.score_undeclared(_PROBES, _all_pass(_PROBES), m, "s")
|
||
assert (row.k, row.status) == (None, gate.NOT_MEASURED)
|
||
|
||
|
||
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 _all_green(tmp_path: Path) -> tuple[list[gate.Row], dict[str, Any]]:
|
||
config = json.loads(json.dumps(_CONFIG))
|
||
config["maf_points"]["points"] = _one_point([6])["points"]
|
||
rows = gate.evaluate(
|
||
rounds_dir=_green_rounds(tmp_path / "rounds"),
|
||
config=config,
|
||
repo_root=_REPO,
|
||
src=_synthetic_src(tmp_path / "src"),
|
||
probe_runner=_all_pass,
|
||
stress_measure=_CLEAN,
|
||
)
|
||
return rows, config
|
||
|
||
|
||
def test_every_failing_row_green_is_exit_zero(tmp_path: Path) -> None:
|
||
rows, _ = _all_green(tmp_path)
|
||
assert [r.status for r in rows[:6]] == [gate.GREEN] * 6, gate.render(rows)
|
||
assert gate.exit_code(rows) == 0
|
||
|
||
|
||
@pytest.mark.parametrize("index", range(6))
|
||
def test_each_failing_row_alone_moves_the_exit_code(tmp_path: Path, index: int) -> None:
|
||
rows, _ = _all_green(tmp_path)
|
||
red = list(rows)
|
||
red[index] = gate.Row(rows[index].key, rows[index].title, 0, 1, gate.RED, "r")
|
||
assert gate.exit_code(red) == 1
|
||
|
||
|
||
def test_the_attestation_is_printed(tmp_path: Path) -> None:
|
||
rows, _ = _all_green(tmp_path)
|
||
assert gate.ATTESTATION in gate.render(rows)
|
||
|
||
|
||
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
|
||
assert _cli("--stress-root", str(tmp_path / "missing")).returncode == 2
|
||
assert _cli("--bundle-root", str(tmp_path / "missing")).returncode == 2
|
||
# m-2: a rounds directory inside the repository that git would commit.
|
||
tracked = _cli("--rounds-dir", str(_REPO / "contexts"))
|
||
assert tracked.returncode == 2 and "gitignored" in tracked.stderr
|
||
help_text = _cli("--help").stdout
|
||
assert "report.kept.md" in help_text and "given_at" in help_text and "outbox" in help_text
|
||
assert _ATTEST_FILE in help_text and "FORM OK" 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
|
||
assert payload["attestation"] == gate.ATTESTATION
|
||
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)
|
||
# Probes green since row 6; stress round 6 predates the rule (or is absent) -> never green.
|
||
assert rows["undeclared"]["status"] == gate.NOT_MEASURED
|
||
assert rows["undeclared"]["k"] is None
|
||
assert rows["named"]["failing"] is False
|