Kjell Tore Guttormsen
9825b2677c
An independent review made rows 1, 2 and 4 green from a handwritten directory in a minute, and 10 of 20 mutants survived the gate's tests. Rounds now need a new point and their own ids, a timezone-aware given_at in order, and a report the feedback was given on; every outcome must name a run whose own coverage confirms (a)-(d), the feedback must fall between the two runs, and a NOK change under 1 % is noise. Row 4 counts content lines kept unchanged and in order, shows the expert's additions, and calls a byte-identical copy untouched unless round 3 acknowledges it. Row 6 counts the runs' own proposals. Types 3 and 7 are proven through the real flags with the action in the result (still 3 of 8). The contract numbers and the evidence register are pinned to their source. Every run prints that rows 1-2 cannot prove who wrote the feedback. A rounds directory inside the repo that git would commit, and a missing stress or bundle root, are usage errors. The review's 20 mutants, re-run: 20 of 20 killed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
793 lines
32 KiB
Python
793 lines
32 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 subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from portfolio_optimiser.evals import v1_gate as gate
|
||
from portfolio_optimiser.validator import UNSUPPORTED_REASON
|
||
|
||
_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 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 _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: an outbox holding the run's own coverage,
|
||
stamped at a fixed time (the run's time is the coverage file's)."""
|
||
run_id = f"r{round_dir.name}"
|
||
outbox = round_dir.parent / "runs" / run_id
|
||
rendered = [_coverage_row(r) for r in rows] if coverage is None else coverage
|
||
_write(outbox / f"{run_id}-coverage.json", {"rows": rendered, "stop_reason": ""})
|
||
stamp = _T0 + (int(round_dir.name) * 20 if at is None else at)
|
||
os.utime(outbox / f"{run_id}-coverage.json", (stamp, stamp))
|
||
_write(
|
||
round_dir / "outcome.json",
|
||
{
|
||
"run_id": run_id,
|
||
"outbox": f"../runs/{run_id}",
|
||
"approaches": rows,
|
||
"removed": list(removed),
|
||
},
|
||
)
|
||
|
||
|
||
def _green_rounds(root: Path) -> Path:
|
||
"""Three traced rounds, each run after its feedback, and a round 3 report the expert kept
|
||
whole while adding a line of their own."""
|
||
_outcome(root / "0", [_row("a1", False, None, stage="stage0-baseline")])
|
||
_write(root / "0" / "report.md", "# Rapport 0\n\nlinje\n")
|
||
for n in (1, 2, 3):
|
||
_feedback(root / str(n), (f"f{n}", 1, f"Tallet for linje {n} er feil, bruk kontrakten."))
|
||
_outcome(root / str(n), [_row("a1", True, 1000.0 * n, f"f{n}")])
|
||
_write(root / str(n) / "report.md", f"# Rapport {n}\n\nlinje\n")
|
||
_write(root / "3" / "report.kept.md", "# Rapport 3\n\nlinje\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 / "runs" / "r1" / "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)
|
||
|
||
|
||
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 = Path("~/repos/vegnormal-okf/build/ferdig").expanduser()
|
||
if not root.is_dir() or not bundles.is_dir():
|
||
m = gate.measure_stress(evidence, _REPO, tmp_path / "absent", bundles)
|
||
assert m.missing and m.validated == 0
|
||
pytest.skip(f"stress artefacts not mounted ({root}, {bundles})")
|
||
m = gate.measure_stress(evidence, _REPO, root, bundles)
|
||
if m.missing:
|
||
# The mount belongs to another repository and can be mid-rebuild; the gate then says
|
||
# "ikke målt", which test_row6_missing_artefacts_are_never_zero already pins.
|
||
assert m.validated == 0
|
||
pytest.skip(f"stress artefacts not judgeable right now: {m.missing}")
|
||
# 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
|
||
|
||
|
||
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
|