1473 lines
63 KiB
Python
1473 lines
63 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 datetime import datetime, timezone
|
||
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 reads FORM OK 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; since M-7 that form is as far as row 1 can get without the
|
||
operator's attestation. The control that the row can still go green is ``_green_rounds``,
|
||
which carries a whole run family AND an attestation (``test_row2_a_traced_change_counts``)."""
|
||
root = _forged_rounds(tmp_path)
|
||
assert gate.score_rounds(root, 3, _AI).status == gate.FORM_OK
|
||
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. Measured as BEHAVIOUR for the whole
|
||
package (18.09, PM): every entry point runs against an unattested but otherwise consistent
|
||
round directory, and none may leave an attestation behind. A grep for the file name reads a
|
||
name, and a writer called from ``render`` or the command line slipped past it."""
|
||
root = _unattested(_green_rounds(tmp_path / "rounds"))
|
||
config = json.loads(json.dumps(_CONFIG))
|
||
config["maf_points"]["points"] = _one_point([6])["points"]
|
||
entries: dict[str, Any] = {
|
||
"score_rounds": lambda: gate.score_rounds(root, 3, _AI),
|
||
"score_changes": lambda: gate.score_changes(root, 3, _AI),
|
||
"score_kept": lambda: gate.score_kept(root, 0.8, _AI),
|
||
"read_attestation": lambda: [gate.read_attestation(root / str(n)) for n in range(4)],
|
||
"evaluate+render": lambda: gate.render(
|
||
gate.evaluate(
|
||
rounds_dir=root,
|
||
config=config,
|
||
repo_root=_REPO,
|
||
src=_synthetic_src(tmp_path / "src"),
|
||
probe_runner=_all_pass,
|
||
stress_measure=_CLEAN,
|
||
)
|
||
),
|
||
}
|
||
for name, entry in entries.items():
|
||
entry()
|
||
assert list(root.rglob(_ATTEST_FILE)) == [], name
|
||
for flags in ((), ("--json",)):
|
||
proc = _cli("--rounds-dir", str(root), *flags)
|
||
assert proc.returncode == 1, proc.stderr
|
||
assert list(root.rglob(_ATTEST_FILE)) == [], ("cli", flags)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# M-8 — what the attestation's DATE and FORM may say (PM checkpoint 18.09 of 1b48124)
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
_NOW = datetime(2026, 5, 29, 12, 0, tzinfo=timezone.utc)
|
||
|
||
|
||
def _both_rows(root: Path, now: datetime | None = None) -> list[gate.Row]:
|
||
return [gate.score_rounds(root, 3, _AI, now=now), gate.score_changes(root, 3, _AI, now=now)]
|
||
|
||
|
||
def test_m8_the_baseline_stays_green_on_the_injected_clock(tmp_path: Path) -> None:
|
||
"""Positive control for everything below: the same tree, the clock on the fixture day."""
|
||
rows = _both_rows(_green_rounds(tmp_path), _NOW)
|
||
assert [(r.k, r.status) for r in rows] == [(3, gate.GREEN)] * 2, rows
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"given",
|
||
["2026-05-30", "3000-01-01", "2026-05-29T23:00:00-05:00"],
|
||
)
|
||
def test_m8_an_attestation_dated_in_the_future_is_red(tmp_path: Path, given: str) -> None:
|
||
"""Finding 1: ``dato: 3000-01-01`` on every round read GREEN — the check only asked whether
|
||
the date was before the run. The last case is a future INSTANT written in a west-of-UTC
|
||
offset, so its wall-clock date is still today."""
|
||
root = _green_rounds(tmp_path)
|
||
_attest(root / "1", on=given)
|
||
for row in _both_rows(root, _NOW):
|
||
assert (row.k, row.status) == (2, gate.RED), (given, row.exceptions)
|
||
assert any("i framtiden" in x for x in row.exceptions), row.exceptions
|
||
|
||
|
||
def test_m8_the_clock_is_a_parameter_and_the_default_is_the_real_one(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_attest(root / "1", on="3000-01-01")
|
||
assert gate.score_rounds(root, 3, _AI).status == gate.RED # the real clock, no injection
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"given",
|
||
["2026-05-29T00:00:00+14:00", "2026-05-28T10:00:00+00:00", "2026-05-28T20:26:00+00:00"],
|
||
)
|
||
def test_m8_a_timezone_cannot_date_an_attestation_before_the_run(
|
||
tmp_path: Path, given: str
|
||
) -> None:
|
||
"""Finding 2: ``2026-05-29T00:00:00+14:00`` is 2026-05-28T10:00Z, ten hours BEFORE the run
|
||
(20:26Z), and read GREEN because only the wall-clock date was compared. A date that carries a
|
||
time is compared as the instant it is."""
|
||
root = _green_rounds(tmp_path)
|
||
_attest(root / "1", on=given)
|
||
for row in _both_rows(root, _NOW):
|
||
assert (row.k, row.status) == (2, gate.RED), (given, row.exceptions)
|
||
assert any("før kjøringen" in x for x in row.exceptions), row.exceptions
|
||
|
||
|
||
def test_m8_a_time_without_an_offset_is_ambiguous_and_red(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_attest(root / "1", on="2026-05-29T10:00:00")
|
||
row = gate.score_rounds(root, 3, _AI, now=_NOW)
|
||
assert (row.k, row.status) == (2, gate.RED), row.exceptions
|
||
assert any("tidssone" in x for x in row.exceptions), row.exceptions
|
||
|
||
|
||
@pytest.mark.parametrize("given", ["2026-05-29T08:00:00+02:00", "2026-05-29T00:00:00+00:00"])
|
||
def test_m8_an_instant_after_the_run_and_before_now_is_green(tmp_path: Path, given: str) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_attest(root / "1", on=given)
|
||
for row in _both_rows(root, _NOW):
|
||
assert (row.k, row.status) == (3, gate.GREEN), (given, row.exceptions)
|
||
|
||
|
||
def test_m8_a_plain_date_keeps_the_same_day_rule(tmp_path: Path) -> None:
|
||
"""Same day as the run, and same day as ``now``, cannot be told apart from a date alone — so
|
||
it passes, and the contract text says so."""
|
||
root = _green_rounds(tmp_path)
|
||
_attest(root / "1", on="2026-05-28") # the run's own day
|
||
assert gate.score_rounds(root, 3, _AI, now=_NOW).status == gate.GREEN
|
||
_attest(root / "1", on="2026-05-29") # today
|
||
assert gate.score_rounds(root, 3, _AI, now=_NOW).status == gate.GREEN
|
||
assert "samme dag" in gate.ATTEST_RULE
|
||
|
||
|
||
_BODY = {"runde": "runde: 1", "kjøring": "kjøring: r1", "dato": "dato: 2026-05-29"}
|
||
|
||
|
||
@pytest.mark.parametrize("key", ["runde", "kjøring", "dato"])
|
||
@pytest.mark.parametrize("first_is_right", [True, False])
|
||
def test_m8_a_key_written_twice_is_red_in_either_order(
|
||
tmp_path: Path, key: str, first_is_right: bool
|
||
) -> None:
|
||
"""Finding 3: ``runde: 1`` then ``runde: 2`` was GREEN and the reverse RED — first-wins by
|
||
accident, and a last-wins reading survived the whole suite. Two answers to one question are
|
||
not an answer, whichever comes first."""
|
||
root = _green_rounds(tmp_path)
|
||
wrong = {"runde": "runde: 2", "kjøring": "kjøring: r9", "dato": "dato: 2020-01-01"}[key]
|
||
pair = [_BODY[key], wrong] if first_is_right else [wrong, _BODY[key]]
|
||
lines = [x for k, x in _BODY.items() if k != key] + pair
|
||
_write(root / "1" / _ATTEST_FILE, "\n".join(lines) + "\n")
|
||
for row in _both_rows(root, _NOW):
|
||
assert (row.k, row.status) == (2, gate.RED), row.exceptions
|
||
assert any("to ganger" in x for x in row.exceptions), row.exceptions
|
||
|
||
|
||
def test_m8_even_the_same_value_twice_is_red(tmp_path: Path) -> None:
|
||
root = _green_rounds(tmp_path)
|
||
_write(root / "1" / _ATTEST_FILE, "\n".join([*_BODY.values(), _BODY["runde"]]) + "\n")
|
||
assert gate.score_rounds(root, 3, _AI, now=_NOW).status == gate.RED
|
||
|
||
|
||
def test_m8_a_byte_order_mark_is_tolerated(tmp_path: Path) -> None:
|
||
"""Finding 5a: a real file from an ordinary editor may start with a BOM, which made the first
|
||
key read as a different word and the round RED. The person is not wrong; the gate tolerates
|
||
it."""
|
||
root = _green_rounds(tmp_path)
|
||
text = "\n".join(_BODY.values()) + "\n"
|
||
(root / "1" / _ATTEST_FILE).write_bytes(b"\xef\xbb\xbf" + text.encode("utf-8"))
|
||
for row in _both_rows(root, _NOW):
|
||
assert (row.k, row.status) == (3, gate.GREEN), row.exceptions
|
||
|
||
|
||
def test_m8_an_attestation_that_is_a_directory_is_red_not_missing(tmp_path: Path) -> None:
|
||
"""Finding 5b: a directory named like the file is something PUT there — not the same as
|
||
nobody having confirmed yet."""
|
||
root = _green_rounds(tmp_path)
|
||
(root / "1" / _ATTEST_FILE).unlink()
|
||
(root / "1" / _ATTEST_FILE).mkdir()
|
||
for row in _both_rows(root, _NOW):
|
||
assert (row.k, row.status) == (2, gate.RED), row.exceptions
|
||
assert any("vanlig fil" in x for x in row.exceptions), row.exceptions
|
||
|
||
|
||
@pytest.mark.parametrize("target", ["valid", "dangling"])
|
||
def test_m8_an_attestation_that_is_a_symlink_is_red(tmp_path: Path, target: str) -> None:
|
||
"""Finding 5c: F4 refuses a linked outbox, and ``is_file()`` followed the same link here. What
|
||
a forger gains is nothing (the content binds), but the asymmetry was unmeasured."""
|
||
root = _green_rounds(tmp_path / "rounds")
|
||
elsewhere = tmp_path / "elsewhere.txt"
|
||
if target == "valid":
|
||
elsewhere.write_text("\n".join(_BODY.values()) + "\n", encoding="utf-8")
|
||
(root / "1" / _ATTEST_FILE).unlink()
|
||
(root / "1" / _ATTEST_FILE).symlink_to(elsewhere)
|
||
for row in _both_rows(root, _NOW):
|
||
assert (row.k, row.status) == (2, gate.RED), (target, row.exceptions)
|
||
assert any("lenke" in x for x in row.exceptions), row.exceptions
|
||
|
||
|
||
def test_m8_a_hard_link_is_a_declared_limit_not_a_rule(tmp_path: Path) -> None:
|
||
"""Chosen 18.09: a hard link is NOT refused, and the output says so. Nothing in the tree marks
|
||
where it points, ``cp -l`` and some backup tools make them innocently, and refusing would gain
|
||
nothing — the attestation's content binds round, run and date either way."""
|
||
root = _green_rounds(tmp_path / "rounds")
|
||
outside = tmp_path / "outside.txt"
|
||
outside.write_text("\n".join(_BODY.values()) + "\n", encoding="utf-8")
|
||
(root / "1" / _ATTEST_FILE).unlink()
|
||
os.link(outside, root / "1" / _ATTEST_FILE)
|
||
assert gate.score_rounds(root, 3, _AI, now=_NOW).status == gate.GREEN
|
||
assert "hardlenke" in gate.ATTEST_RULE
|
||
|
||
|
||
def test_m8_an_attestation_that_is_not_utf8_is_red_because_it_is_unreadable(
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""Finding 6: the ``except`` arm on the read was uncovered — a whole-file garbage input goes
|
||
RED anyway through the missing keys, so a lenient decode (``errors="ignore"``) survived. Here
|
||
the three keyed lines are perfect and ONE stray byte follows: only strictness can refuse it."""
|
||
root = _green_rounds(tmp_path)
|
||
good = ("\n".join(_BODY.values()) + "\n").encode("utf-8")
|
||
(root / "1" / _ATTEST_FILE).write_bytes(good + b"merknad \xff\xfe\n")
|
||
for row in _both_rows(root, _NOW):
|
||
assert (row.k, row.status) == (2, gate.RED), row.exceptions
|
||
assert any("uleselig" in x for x in row.exceptions), row.exceptions
|
||
|
||
|
||
def test_m8_the_output_says_the_stress_row_needs_the_untracked_scratchpad() -> None:
|
||
"""Finding 7: row 7's 1 of 20 stands on ``scratchpad/`` being present. Without it the row is
|
||
NOT MEASURED, and that dependency is a fact about the checkout, not about the product."""
|
||
assert "scratchpad" in gate.STRESS_DEPENDENCY and "IKKE MÅLT" in gate.STRESS_DEPENDENCY
|
||
rows = gate.evaluate(
|
||
rounds_dir=Path("/nonexistent"),
|
||
config=_CONFIG,
|
||
repo_root=_REPO,
|
||
probe_runner=_all_pass,
|
||
stress_measure=_CLEAN,
|
||
)
|
||
assert gate.STRESS_DEPENDENCY in gate.render(rows)
|
||
|
||
|
||
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
|