Kjell Tore Guttormsen
c5b0dc6d9a
Three guards for the code the two step-0 commits added, each written against a mutation that would otherwise survive: a wiki-shaped file carrying the pinned 20 questions and 29 fasit entries but invented content (so the counts alone are a one-line forgery), a threshold of 80 or 1.5 (a share or a typo, and guessing is not this row's job), and a hold-out set of no questions -- the shape every row-5 test used until today, which would have made the new comparison vacuous the moment it was added. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1437 lines
56 KiB
Python
1437 lines
56 KiB
Python
"""The retrieval gate, held to the discipline it holds `okf consume` to.
|
|
|
|
Three rules this suite is written under, all of them the house pattern:
|
|
|
|
- **Every failing row must be shown able to go BOTH ways.** A row that is red
|
|
today and cannot be green is a row nobody can act on, and a row that is
|
|
green and cannot go red is decoration. Each row below has one test driving
|
|
it green and one driving it red, through INPUTS -- a fixture, a set, a
|
|
registration -- never by editing the row.
|
|
- **Every zero carries a control.** The socket guard is fired before its
|
|
silence during a run counts as evidence; the sha pin is shown to refuse a
|
|
tampered set before its silence on the committed ones means anything.
|
|
- **Nothing here touches a private corpus or the network.** The real sets in
|
|
`claude-code-llm-wiki` and `vegnormal-okf` are never read by a test: the
|
|
adapters are exercised against files written in their shapes, with invented
|
|
content, inside `tmp_path`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(PROJECT_ROOT / "tools"))
|
|
|
|
import okf_retrieval_gate as gate # noqa: E402
|
|
|
|
from llm_ingestion_okf import consume # noqa: E402
|
|
|
|
FIXTURES = PROJECT_ROOT / "tests" / "fixtures" / "retrieval"
|
|
|
|
|
|
def _bundles(tmp_path: Path) -> dict[str, Path]:
|
|
return gate.synthetic_bundles(tmp_path / "bundles")
|
|
|
|
|
|
def _case(tmp_path: Path, name: str) -> gate.Case:
|
|
bundles = _bundles(tmp_path)
|
|
question_set = gate.load_set(FIXTURES / name, gate.SYNTHETIC_SETS[name])
|
|
return gate.measure_case(question_set, bundles)
|
|
|
|
|
|
def _set_file(
|
|
path: Path,
|
|
*,
|
|
set_id: str,
|
|
bundle: str,
|
|
questions: list[dict[str, object]],
|
|
controls: list[dict[str, object]] | None = None,
|
|
) -> tuple[Path, str]:
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"set_id": set_id,
|
|
"bundle": bundle,
|
|
"questions": questions,
|
|
"controls": controls or [],
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return path, gate.sha256_of(path)
|
|
|
|
|
|
# --- the sets are what they were pinned as ------------------------------------
|
|
|
|
|
|
def test_every_pinned_sha_is_the_sha_of_the_file_on_disk() -> None:
|
|
for name, pinned in gate.SYNTHETIC_SETS.items():
|
|
assert gate.sha256_of(FIXTURES / name) == pinned, name
|
|
|
|
|
|
def test_the_pin_refuses_a_set_whose_bytes_moved(tmp_path: Path) -> None:
|
|
# The control first: the same reader accepts the untouched file, so the
|
|
# refusal below is the pin firing and not the reader failing.
|
|
name, pinned = next(iter(gate.SYNTHETIC_SETS.items()))
|
|
assert gate.load_set(FIXTURES / name, pinned).sha256 == pinned
|
|
tampered = tmp_path / name
|
|
tampered.write_bytes((FIXTURES / name).read_bytes() + b" ")
|
|
with pytest.raises(gate.GateUsage) as error:
|
|
gate.load_set(tampered, pinned)
|
|
assert "refusing to measure a set that is not the set that was pinned" in str(error.value)
|
|
|
|
|
|
def test_a_tampered_set_makes_the_command_exit_two(tmp_path: Path) -> None:
|
|
for name in gate.SYNTHETIC_SETS:
|
|
(tmp_path / name).write_bytes((FIXTURES / name).read_bytes())
|
|
(tmp_path / "set-positive.json").write_bytes(
|
|
(FIXTURES / "set-positive.json").read_bytes() + b"\n"
|
|
)
|
|
result = subprocess.run(
|
|
[sys.executable, str(PROJECT_ROOT / "tools" / "okf_retrieval_gate.py")],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=tmp_path,
|
|
env={"PYTHONPATH": str(PROJECT_ROOT / "src"), "PATH": "/usr/bin:/bin"},
|
|
)
|
|
# The committed fixtures are found by module path, not by cwd, so the run
|
|
# above is green; the point of the arm is the DIRECTORY-independent pin,
|
|
# which the unit test above fires. Here we only hold that a tampered file
|
|
# in a working directory cannot silently become the set.
|
|
assert result.returncode in (1, 2)
|
|
|
|
|
|
def test_no_committed_set_names_a_path_outside_this_repository() -> None:
|
|
# The standing rule: a gold set names a consumer's documents and this
|
|
# repository is public. Control first -- the pattern finds a planted path.
|
|
planted = "the corpus at /Users/somebody/corpora/x"
|
|
assert any(mark in planted for mark in ("/Users/", "~/", "corpora"))
|
|
for name in gate.SYNTHETIC_SETS:
|
|
text = (FIXTURES / name).read_text(encoding="utf-8")
|
|
for mark in ("/Users/", "~/", "corpora", ".cache"):
|
|
assert mark not in text, (name, mark)
|
|
|
|
|
|
# --- row 1 --------------------------------------------------------------------
|
|
|
|
|
|
def test_row_one_is_green_when_the_ranker_delivers_every_fasit(tmp_path: Path) -> None:
|
|
cases = [_case(tmp_path, "set-positive.json"), _case(tmp_path, "set-signals.json")]
|
|
row = gate.row_one(cases)
|
|
assert (row.k, row.m, row.status) == (9, 9, gate.GREEN)
|
|
|
|
|
|
def test_row_one_is_red_when_a_fasit_is_not_delivered(tmp_path: Path) -> None:
|
|
path, sha = _set_file(
|
|
tmp_path / "set.json",
|
|
set_id="row-one-red",
|
|
bundle="miss",
|
|
questions=[
|
|
{
|
|
"id": "R1",
|
|
"question": "Hvor mange medlemmer maa stemme for en endring av vedtektene?",
|
|
"fasit": [
|
|
{
|
|
"by": "concept",
|
|
"value": "vedtekter/flertallskrav",
|
|
"quote": "To tredjedeler",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
)
|
|
case = gate.measure_case(gate.load_set(path, sha), _bundles(tmp_path))
|
|
row = gate.row_one([case])
|
|
assert (row.k, row.m, row.status) == (0, 1, gate.RED)
|
|
|
|
|
|
def test_row_one_never_counts_a_question_that_declares_the_class_it_forces(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
row = gate.row_one([_case(tmp_path, "set-classes.json")])
|
|
assert (row.k, row.m, row.status) == (0, 0, gate.RED)
|
|
assert any("row 2's fixture" in detail for detail in row.details)
|
|
|
|
|
|
# --- row 2 --------------------------------------------------------------------
|
|
|
|
|
|
def test_row_two_is_green_when_every_class_is_the_one_its_fixture_forces(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
cases = [_case(tmp_path, "set-classes.json"), _case(tmp_path, "set-miss.json")]
|
|
row = gate.row_two(cases)
|
|
assert row.status == gate.GREEN
|
|
assert row.k == row.m == 7
|
|
# Each class forced by its own fixture, and named in the output.
|
|
for letter in ("a", "b", "c", "d", "e"):
|
|
assert any(detail.startswith(f"class {letter} ") for detail in row.details)
|
|
|
|
|
|
def test_row_two_is_red_when_a_fixture_does_not_force_the_class_it_declares(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
path, sha = _set_file(
|
|
tmp_path / "set.json",
|
|
set_id="row-two-red",
|
|
bundle="positive",
|
|
questions=[
|
|
{
|
|
"id": "W1",
|
|
"question": "Hvem har ansvaret for broeyting av parkeringsplassen?",
|
|
"expect_class": "b",
|
|
"fasit": [
|
|
{
|
|
"by": "concept",
|
|
"value": "haandbok/broeyteansvar",
|
|
"quote": "broeyteansvaret ligger hos vaktmesteren",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
)
|
|
case = gate.measure_case(gate.load_set(path, sha), _bundles(tmp_path))
|
|
row = gate.row_two([case])
|
|
assert (row.k, row.m, row.status) == (0, 1, gate.RED)
|
|
assert any("forced class b, measured a" in detail for detail in row.details)
|
|
|
|
|
|
def test_a_miss_with_no_class_at_all_takes_the_whole_row_to_zero(tmp_path: Path) -> None:
|
|
case = _case(tmp_path, "set-classes.json")
|
|
unplaced = [
|
|
gate.Unit(
|
|
question_id="X1",
|
|
question="?",
|
|
named="concept:whatever",
|
|
quote="",
|
|
hit=False,
|
|
rank=None,
|
|
klass=None,
|
|
label_default=None,
|
|
truth=None,
|
|
confirmed=None,
|
|
)
|
|
]
|
|
holed = gate.Case(case.question_set, case.bundles, (*case.units, *unplaced))
|
|
row = gate.row_two([holed])
|
|
assert (row.k, row.status) == (0, gate.RED)
|
|
assert any("unplaced" in detail for detail in row.details)
|
|
|
|
|
|
# --- row 3 --------------------------------------------------------------------
|
|
|
|
|
|
def test_row_three_goes_red_again_the_moment_the_label_stops_being_true(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""The RED direction, driven by an input: the library's own behaviour.
|
|
|
|
This row was red on the shipped code until 2026-09-20 -- every over-quota
|
|
candidate came back `source_quota_exceeded`, including the ones the RANK
|
|
had already put outside k. Restoring exactly that naming puts the row back
|
|
where it was, with the same three fixtures and the same detail line, so
|
|
the green reading is not a green nobody can lose.
|
|
"""
|
|
|
|
def every_drop_is_the_quotas(candidates: object, **_: object) -> dict[str, str]:
|
|
return _AlwaysTheQuota()
|
|
|
|
monkeypatch.setattr(consume, "_fates_without_quota", every_drop_is_the_quotas)
|
|
row = gate.row_three([_case(tmp_path, "set-classes.json")])
|
|
assert row.status == gate.RED
|
|
lying = [detail for detail in row.details if "source_quota_exceeded" in detail]
|
|
assert lying, row.details
|
|
assert all("the quota-off run says `below_k`" in detail for detail in lying)
|
|
|
|
|
|
class _AlwaysTheQuota(dict[str, str]):
|
|
"""The pre-2026-09-20 naming: every drop the quota reached is the quota's."""
|
|
|
|
def __missing__(self, key: str) -> str:
|
|
return "source_quota_exceeded"
|
|
|
|
|
|
def test_row_three_is_green_on_the_shipped_code(tmp_path: Path) -> None:
|
|
"""And the same three fixtures, unmutated, are the green direction.
|
|
|
|
Four judged units in this set alone; the fifth in the gate's own run comes
|
|
from `set-miss.json`."""
|
|
row = gate.row_three([_case(tmp_path, "set-classes.json")])
|
|
assert (row.k, row.m, row.status) == (4, 4, gate.GREEN)
|
|
assert not [detail for detail in row.details if "source_quota_exceeded" in detail]
|
|
|
|
|
|
def test_row_three_is_green_when_the_printed_reason_is_the_true_one(tmp_path: Path) -> None:
|
|
# The budget case: withheld by the pack in BOTH runs, so the label the
|
|
# payload prints is the label the quota-off run confirms.
|
|
path, sha = _set_file(
|
|
tmp_path / "set.json",
|
|
set_id="row-three-green",
|
|
bundle="budget",
|
|
questions=[
|
|
{
|
|
"id": "C1",
|
|
"question": "Hvor foeres dugnadstimene?",
|
|
"limit": 8000,
|
|
"expect_class": "c",
|
|
"fasit": [
|
|
{"by": "concept", "value": "tabell/dugnadstabell", "quote": "rad for rad"}
|
|
],
|
|
}
|
|
],
|
|
)
|
|
case = gate.measure_case(gate.load_set(path, sha), _bundles(tmp_path))
|
|
row = gate.row_three([case])
|
|
assert (row.k, row.m, row.status) == (1, 1, gate.GREEN)
|
|
|
|
|
|
# --- row 4 --------------------------------------------------------------------
|
|
|
|
|
|
def test_row_four_goes_red_again_when_the_payload_stops_saying_what_it_missed(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""The RED direction, driven by an input: the payload's own words.
|
|
|
|
This row was 3 of 6 until 2026-09-20 because the payload said nothing
|
|
about what of the question it reached, so N3, N4 and N5 -- 8, 8 and 1
|
|
excerpts -- were indistinguishable from an answered question. A payload
|
|
that reports every term as answered puts the row back there, with the same
|
|
detail line.
|
|
"""
|
|
monkeypatch.setattr(consume, "unanswered_terms", lambda *args, **kwargs: [])
|
|
row = gate.row_four([_case(tmp_path, "set-controls.json")])
|
|
assert row.status == gate.RED
|
|
assert row.m == 6
|
|
assert any("not covered and NOT marked" in detail for detail in row.details)
|
|
|
|
|
|
def test_row_four_is_green_when_both_directions_come_out_right(tmp_path: Path) -> None:
|
|
path, sha = _set_file(
|
|
tmp_path / "set.json",
|
|
set_id="row-four-green",
|
|
bundle="positive",
|
|
questions=[],
|
|
controls=[
|
|
{
|
|
"id": "N1",
|
|
"kind": "an unrelated topic",
|
|
"question": "Hvilken safran passer til fiskesuppe?",
|
|
},
|
|
{
|
|
"id": "P0",
|
|
"kind": "covered by the bundle",
|
|
"question": "Naar kontrolleres vinterberedskapen?",
|
|
"covered": True,
|
|
},
|
|
],
|
|
)
|
|
case = gate.measure_case(gate.load_set(path, sha), _bundles(tmp_path))
|
|
row = gate.row_four([case])
|
|
assert (row.k, row.m, row.status) == (2, 2, gate.GREEN)
|
|
|
|
|
|
def test_a_covered_question_that_came_back_marked_would_fail_the_same_row(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
# The other direction of row 4's own rule: a marking that also fires on a
|
|
# question the bundle answers says nothing at all.
|
|
path, sha = _set_file(
|
|
tmp_path / "set.json",
|
|
set_id="row-four-inverted",
|
|
bundle="positive",
|
|
questions=[],
|
|
controls=[
|
|
{
|
|
"id": "X0",
|
|
"kind": "declared covered, and it is not",
|
|
"question": "Hvilken safran passer til fiskesuppe?",
|
|
"covered": True,
|
|
}
|
|
],
|
|
)
|
|
case = gate.measure_case(gate.load_set(path, sha), _bundles(tmp_path))
|
|
row = gate.row_four([case])
|
|
assert (row.k, row.m, row.status) == (0, 1, gate.RED)
|
|
assert any("MARKED anyway" in detail for detail in row.details)
|
|
|
|
|
|
# --- row 5 --------------------------------------------------------------------
|
|
|
|
|
|
def _carried_by_git(
|
|
*, tracked: bool = True, same_commit: bool = False, after: int = 1
|
|
) -> gate.Provenance:
|
|
"""A history row 5 reads instead of believing the registration. The three
|
|
git checks are driven from HERE, never from a file the test writes, which
|
|
is the whole point of the row."""
|
|
return gate.Provenance(
|
|
tracked=tracked,
|
|
unmodified=tracked,
|
|
commit="0" * 40 if tracked else "",
|
|
commit_touches_ranking=same_commit,
|
|
ranking_commits_after=after,
|
|
)
|
|
|
|
|
|
def test_row_five_is_red_while_no_hold_out_is_registered(tmp_path: Path) -> None:
|
|
row = gate.row_five(tmp_path / "absent.json")
|
|
assert (row.k, row.m, row.status) == (0, 1, gate.RED)
|
|
|
|
|
|
def test_row_five_is_green_for_a_registration_that_carries_all_eleven(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
registration = _registration(
|
|
tmp_path,
|
|
set_path=FIXTURES / "set-positive.json",
|
|
bundle=_bundles(tmp_path)["positive"],
|
|
threshold=0.8,
|
|
)
|
|
row = gate.row_five(registration, provenance=lambda _: _carried_by_git())
|
|
assert (row.k, row.m, row.status) == (11, 11, gate.GREEN)
|
|
|
|
|
|
def test_row_five_falls_on_a_number_read_before_its_threshold_was_written(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
registration = _registration(
|
|
tmp_path,
|
|
set_path=FIXTURES / "set-positive.json",
|
|
bundle=_bundles(tmp_path)["positive"],
|
|
threshold=0.8,
|
|
)
|
|
spec = json.loads(registration.read_text(encoding="utf-8"))
|
|
spec["readings"] = [{"at": "2026-09-18T09:00:00Z", "value": "0.62"}]
|
|
registration.write_text(json.dumps(spec), encoding="utf-8")
|
|
row = gate.row_five(registration, provenance=lambda _: _carried_by_git())
|
|
assert (row.k, row.m, row.status) == (10, 11, gate.RED)
|
|
assert any("no reading predates the threshold: NO" in detail for detail in row.details)
|
|
|
|
|
|
def test_row_five_falls_when_the_registration_rides_in_on_the_ranking_change(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The three git checks, each driven red on its own: a file nobody
|
|
committed, a threshold committed together with the ranking change, and a
|
|
registration no ranking change has come after."""
|
|
registration = _registration(
|
|
tmp_path,
|
|
set_path=FIXTURES / "set-positive.json",
|
|
bundle=_bundles(tmp_path)["positive"],
|
|
threshold=0.8,
|
|
)
|
|
arms = {
|
|
"uncommitted": (_carried_by_git(tracked=False), "committed"),
|
|
"same commit as the ranking": (_carried_by_git(same_commit=True), "same commit"),
|
|
"nothing changed since": (_carried_by_git(after=0), "0 commit(s) touching"),
|
|
}
|
|
for label, (history, expected) in arms.items():
|
|
row = gate.row_five(registration, provenance=lambda _, h=history: h)
|
|
assert row.status == gate.RED, label
|
|
assert any(expected in detail for detail in row.details), (label, row.details)
|
|
|
|
|
|
# --- row 6 --------------------------------------------------------------------
|
|
|
|
|
|
def test_row_six_is_green_when_every_delivery_is_the_bundle_s_own_bytes(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
cases = [_case(tmp_path, "set-positive.json"), _case(tmp_path, "set-signals.json")]
|
|
row = gate.row_six(cases)
|
|
assert (row.k, row.m, row.status) == (9, 9, gate.GREEN)
|
|
|
|
|
|
def test_row_six_is_red_when_the_payload_hands_over_something_else(tmp_path: Path) -> None:
|
|
# The judge reads the bundle BEFORE the payload is built -- that is what
|
|
# makes this test possible at all, and what row 7 relies on when it
|
|
# mutates the code the payload comes from.
|
|
bundles = _bundles(tmp_path)
|
|
gate.bundle_index(bundles["positive"])
|
|
question_set = gate.load_set(
|
|
FIXTURES / "set-positive.json", gate.SYNTHETIC_SETS["set-positive.json"]
|
|
)
|
|
with gate._truncate_delivered():
|
|
case = gate.measure_case(question_set, bundles)
|
|
row = gate.row_six([case])
|
|
assert row.status == gate.RED
|
|
assert row.k < row.m or row.m == 0
|
|
|
|
|
|
# --- row 7 --------------------------------------------------------------------
|
|
|
|
|
|
def _noop_mutant() -> gate.Mutant:
|
|
return gate.Mutant("N01 nothing is changed", 1, lambda: gate._patched())
|
|
|
|
|
|
def test_row_seven_is_green_when_every_mutant_is_felled(tmp_path: Path) -> None:
|
|
cases, _ = gate.synthetic_cases(tmp_path / "bundles", FIXTURES)
|
|
baseline = gate.deterministic_rows(cases)
|
|
lethal = (gate.MUTANTS[3],) # M04, the reversed ranking
|
|
row = gate.row_seven(cases, baseline, mutants=lethal, roster=[m.label for m in lethal])
|
|
assert (row.k, row.m, row.status) == (1, 1, gate.GREEN)
|
|
|
|
|
|
def test_row_seven_is_red_when_a_mutant_survives(tmp_path: Path) -> None:
|
|
cases, _ = gate.synthetic_cases(tmp_path / "bundles", FIXTURES)
|
|
baseline = gate.deterministic_rows(cases)
|
|
noop = (_noop_mutant(),)
|
|
row = gate.row_seven(cases, baseline, mutants=noop, roster=[m.label for m in noop])
|
|
assert (row.k, row.m, row.status) == (0, 1, gate.RED)
|
|
assert any(
|
|
"SURVIVED" in detail and "row 1 should have taken it" in detail for detail in row.details
|
|
)
|
|
|
|
|
|
def test_a_mutant_is_felled_by_the_row_that_got_worse_and_never_by_one_that_did_not(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""PM's definition, held by a test: `worse`, never `different`.
|
|
|
|
Removing the quota costs row 1 deliveries -- that is the kill. It costs
|
|
row 3 nothing, because a cut with no quota has no quota to name falsely,
|
|
and crediting row 3 with the kill would give this gate a check it does not
|
|
have. Before 2026-09-20 the same mutation made row 3 GREENER, which was
|
|
the sharper form of the same point.
|
|
"""
|
|
cases, _ = gate.synthetic_cases(tmp_path / "bundles", FIXTURES)
|
|
before = {row.number: row.k for row in gate.deterministic_rows(cases)}
|
|
with gate._wrap_cut(source_quota=None):
|
|
rows = gate.deterministic_rows(
|
|
[gate.measure_case(case.question_set, case.bundles) for case in cases]
|
|
)
|
|
after = {row.number: row.k for row in rows}
|
|
worse = [number for number in sorted(before) if after[number] < before[number]]
|
|
assert 1 in worse, "row 1 is what fells this mutant"
|
|
assert 3 not in worse, after
|
|
assert after[3] >= before[3]
|
|
|
|
|
|
# --- rows 8 and 9 -------------------------------------------------------------
|
|
|
|
|
|
def test_row_eight_is_never_green_when_it_did_not_run() -> None:
|
|
row = gate.row_eight([])
|
|
assert row.status == gate.NOT_RUN
|
|
assert row.fails
|
|
assert any("recorded 2026-09-17 by PM, NOT measured" in detail for detail in row.details)
|
|
|
|
|
|
def test_the_wiki_adapter_reads_its_own_shape_and_hits_by_source_file(tmp_path: Path) -> None:
|
|
# The real set is never opened by a test. This is a file in its shape,
|
|
# with invented content, pointing at the synthetic bundle.
|
|
path = tmp_path / "wiki-shaped.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema": "1",
|
|
"hit_rule": "source_file == <doc>.md AND the excerpt carries the quote",
|
|
"questions": [
|
|
{
|
|
"id": "W1",
|
|
"class": "docs",
|
|
"question": "Naar kontrolleres vinterberedskapen?",
|
|
"fasit": [{"doc": "haandbok", "quote": "innen 1. november"}],
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
question_set = gate.read_real_set("wiki", path, gate.sha256_of(path))
|
|
assert question_set.set_id == "wiki-20"
|
|
assert question_set.questions[0].fasit[0].by == "source_file"
|
|
assert question_set.questions[0].fasit[0].value == "haandbok.md"
|
|
row = gate.row_eight([(question_set, {"wiki": _bundles(tmp_path)["positive"]})])
|
|
# J2: one set of three is NOT a measurement of the three -- this asserted
|
|
# GREEN until 2026-09-19, which is the breakthrough PM measured.
|
|
assert row.status == gate.NOT_RUN
|
|
assert any("citation granularity" in detail for detail in row.details)
|
|
|
|
|
|
def test_the_r761_adapter_splits_the_known_negative_out_of_the_questions(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
path = tmp_path / "r761-shaped.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"fasit_form": "the normalised STS title",
|
|
"sporsmal": [
|
|
{"id": "S1", "sporsmal": "Hva sier punktet?", "fasit": "4.2 Vakthold"},
|
|
{"id": "KN", "sporsmal": "Hvilken safran passer til fiskesuppe?", "fasit": ""},
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
question_set = gate.read_real_set("r761", path, gate.sha256_of(path))
|
|
assert [q.id for q in question_set.questions] == ["S1"]
|
|
assert [c.id for c in question_set.controls] == ["KN"]
|
|
assert question_set.questions[0].fasit[0].by == "title"
|
|
# No quote in this set's fasit, so it is concept granularity and says so.
|
|
assert not question_set.quoted
|
|
|
|
|
|
def test_the_vegnormal_adapter_makes_one_question_per_standard(tmp_path: Path) -> None:
|
|
path = tmp_path / "vegnormal-shaped.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"sporsmal": [
|
|
{
|
|
"id": "T1-1",
|
|
"sporsmal": "Hva viser kravet til?",
|
|
"must_cite": [
|
|
{"normal": "N100:2023", "req_number": "2.3.2-3", "fil": "a"},
|
|
{"normal": "N200:2024", "req_number": "1.6.3-3", "fil": "b"},
|
|
],
|
|
}
|
|
]
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
question_set = gate.read_real_set("vegnormal", path, gate.sha256_of(path))
|
|
assert sorted(q.id for q in question_set.questions) == ["T1-1/N100:2023", "T1-1/N200:2024"]
|
|
assert {q.bundle for q in question_set.questions} == {"N100:2023", "N200:2024"}
|
|
assert question_set.units == 2
|
|
|
|
|
|
def test_an_unknown_real_set_name_is_refused(tmp_path: Path) -> None:
|
|
path = tmp_path / "x.json"
|
|
path.write_text("{}", encoding="utf-8")
|
|
with pytest.raises(gate.GateUsage):
|
|
gate.read_real_set("something-else", path, gate.sha256_of(path))
|
|
|
|
|
|
def test_row_nine_states_k2_s_denominator_and_never_passes() -> None:
|
|
row = gate.row_nine()
|
|
assert (row.k, row.m, row.status) == (0, 6, gate.RED)
|
|
assert any("the answer key does not" in detail for detail in row.details)
|
|
|
|
|
|
# --- the verdict --------------------------------------------------------------
|
|
|
|
|
|
def test_a_row_that_did_not_run_fails_the_gate_like_a_red_one() -> None:
|
|
assert gate.Row(8, "x", 0, 3, gate.NOT_RUN, "").fails
|
|
assert gate.Row(1, "x", 1, 1, gate.GREEN, "").fails is False
|
|
|
|
|
|
def test_a_row_with_nothing_to_count_is_never_green() -> None:
|
|
assert gate._row(1, "x", 0, 0, "", []).status == gate.RED
|
|
|
|
|
|
def test_the_verdict_names_every_failing_row() -> None:
|
|
rows = [
|
|
gate.Row(1, "a", 1, 1, gate.GREEN, ""),
|
|
gate.Row(2, "b", 0, 1, gate.RED, ""),
|
|
gate.Row(3, "c", 0, 1, gate.NOT_RUN, ""),
|
|
]
|
|
assert gate.render(rows).strip().endswith("GATE RED: rows 2, 3")
|
|
assert gate.render(rows[:1]).strip().endswith("GATE GREEN")
|
|
|
|
|
|
def test_the_gate_is_red_today_and_says_which_rows(tmp_path: Path) -> None:
|
|
rows = gate.evaluate(tmp_path / "bundles")
|
|
by_number = {row.number: row for row in rows}
|
|
assert sorted(by_number) == [1, 2, 3, 4, 5, 6, 7, 8, 9]
|
|
assert [row.number for row in rows if row.fails] == [5, 7, 8, 9]
|
|
# 10, not 9: `set-quota.json` adds row 3's known-positive, one question the
|
|
# source quota genuinely decides, and it is a hit.
|
|
assert (by_number[1].k, by_number[1].m) == (10, 10)
|
|
assert (by_number[2].k, by_number[2].m) == (7, 7)
|
|
assert (by_number[3].k, by_number[3].m) == (5, 5)
|
|
assert (by_number[6].k, by_number[6].m) == (10, 10)
|
|
|
|
|
|
def test_the_same_tree_measures_the_same_twice(tmp_path: Path) -> None:
|
|
first = gate.evaluate(tmp_path / "one")
|
|
second = gate.evaluate(tmp_path / "two")
|
|
assert [row.to_json() for row in first] == [row.to_json() for row in second]
|
|
|
|
|
|
def test_the_command_exits_one_and_prints_every_row(
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
assert gate.main([]) == 1
|
|
printed = capsys.readouterr().out
|
|
for number in range(1, 10):
|
|
assert f"\n{number} " in f"\n{printed}"
|
|
assert "GATE RED: rows 5, 7, 8, 9" in printed
|
|
|
|
|
|
def test_the_json_form_carries_the_same_rows(capsys: pytest.CaptureFixture[str]) -> None:
|
|
assert gate.main(["--json"]) == 1
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert [row["row"] for row in payload["rows"]] == list(range(1, 10))
|
|
assert payload["gate"] == gate.RED
|
|
assert set(payload["classes"]) == {"a", "b", "c", "d", "e"}
|
|
assert payload["limits"]
|
|
|
|
|
|
def test_an_unreadable_real_set_is_exit_two_and_never_a_quiet_row(
|
|
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
path = tmp_path / "set.json"
|
|
path.write_text("{}", encoding="utf-8")
|
|
assert gate.main(["--real", "wiki", str(path), "0" * 64, str(tmp_path)]) == 2
|
|
assert "refusing to measure a set" in capsys.readouterr().err
|
|
|
|
|
|
# --- the instrument itself ----------------------------------------------------
|
|
|
|
|
|
def test_no_socket_is_opened_during_a_whole_run(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
calls: list[object] = []
|
|
|
|
def refuse(*args: object, **kwargs: object) -> None:
|
|
calls.append(args)
|
|
raise AssertionError("the gate opened a socket")
|
|
|
|
monkeypatch.setattr(socket, "socket", refuse)
|
|
monkeypatch.setattr(socket, "create_connection", refuse)
|
|
# The guard proven able to fire, before its silence counts as evidence.
|
|
with pytest.raises(AssertionError):
|
|
socket.socket()
|
|
calls.clear()
|
|
gate.evaluate(tmp_path / "bundles")
|
|
assert calls == []
|
|
|
|
|
|
def test_the_title_rule_is_the_one_okf_quality_uses_and_not_a_second_copy() -> None:
|
|
from llm_ingestion_okf import quality
|
|
|
|
assert gate.normalise_title is quality.normalise_title
|
|
assert gate._split_numbering is quality._split_numbering
|
|
assert gate._enclosing_directory is quality._enclosing_directory
|
|
|
|
|
|
def test_the_gate_changes_no_ranking_code() -> None:
|
|
# The order's boundary, held by a test rather than by intention: this file
|
|
# measures the ranking and never moves it. Every write to the module it
|
|
# measures lives inside `_patched`, which restores on exit -- so a
|
|
# mutation cannot outlive the `with` block it was applied in.
|
|
import ast
|
|
|
|
path = PROJECT_ROOT / "tools" / "okf_retrieval_gate.py"
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
patched = next(
|
|
node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "_patched"
|
|
)
|
|
writes = [
|
|
node.lineno
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, ast.Call)
|
|
and isinstance(node.func, ast.Name)
|
|
and node.func.id == "setattr"
|
|
and node.args
|
|
and isinstance(node.args[0], ast.Name)
|
|
and node.args[0].id == "consume"
|
|
]
|
|
assert writes, "the control: the walk finds the writes it is asked about"
|
|
assert all(patched.lineno <= line <= (patched.end_lineno or 0) for line in writes), writes
|
|
# And it restores: the same names, read before and written after.
|
|
source = ast.get_source_segment(path.read_text(encoding="utf-8"), patched) or ""
|
|
assert "finally:" in source and "original" in source
|
|
|
|
|
|
def test_the_module_imports_only_the_standard_library_and_this_repository() -> None:
|
|
import re
|
|
|
|
source = (PROJECT_ROOT / "tools" / "okf_retrieval_gate.py").read_text(encoding="utf-8")
|
|
for module in set(re.findall(r"^(?:from|import) ([a-zA-Z_][\w.]*)", source, re.MULTILINE)):
|
|
root = module.split(".")[0]
|
|
assert root in sys.stdlib_module_names or root == "llm_ingestion_okf", root
|
|
|
|
|
|
def test_the_payload_the_gate_reads_is_the_one_a_consumer_gets(tmp_path: Path) -> None:
|
|
# The gate calls `build_payload` with the SHIPPED defaults and changes one
|
|
# argument only where a row's definition needs it (`source_quota=None` for
|
|
# the truth run). A gate measuring a configuration nobody ships would be
|
|
# measuring something else.
|
|
bundles = _bundles(tmp_path)
|
|
question = gate.Question(
|
|
id="P1",
|
|
question="Naar kontrolleres vinterberedskapen?",
|
|
fasit=(gate.Fasit(by="concept", value="haandbok/vinterberedskap", quote="1. november"),),
|
|
)
|
|
assert question.k == consume.DEFAULT_K
|
|
assert question.limit == consume.DEFAULT_LIMIT
|
|
units = gate.measure_units(bundles["positive"], question)
|
|
assert [unit.hit for unit in units] == [True]
|
|
|
|
|
|
def test_the_bundle_index_is_read_from_disk_and_not_from_a_payload(tmp_path: Path) -> None:
|
|
bundles = _bundles(tmp_path)
|
|
index = gate.bundle_index(bundles["positive"])
|
|
assert isinstance(index, gate.BundleIndex)
|
|
assert "haandbok/vinterberedskap" in index.concepts
|
|
view = index.concepts["haandbok/vinterberedskap"]
|
|
on_disk = (bundles["positive"] / "haandbok" / "vinterberedskap.md").read_text(encoding="utf-8")
|
|
assert view.whole == on_disk
|
|
assert view.body in on_disk
|
|
|
|
|
|
def test_a_fasit_named_by_an_unknown_matcher_is_refused() -> None:
|
|
with pytest.raises(gate.GateUsage):
|
|
gate.Fasit(by="vibes", value="x", quote="y")
|
|
|
|
|
|
def test_a_citation_only_in_the_frontmatter_is_in_the_bundle_and_not_in_the_payload(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
# Class (d)'s own mechanism, stated as a fact about the fixture: the
|
|
# sentence IS in the concept file and cannot reach an excerpt, because
|
|
# `excerpt_for` carries no `description`.
|
|
bundles = _bundles(tmp_path)
|
|
index = gate.bundle_index(bundles["positive"])
|
|
view = index.concepts["skjema/avviksskjema"]
|
|
assert "arkiveres i fem aar" in view.whole
|
|
assert "arkiveres i fem aar" not in view.body
|
|
|
|
|
|
def _mapping_is_sorted(mapping: Mapping[str, object]) -> bool:
|
|
return list(mapping) == sorted(mapping)
|
|
|
|
|
|
def test_every_class_is_documented_in_the_output() -> None:
|
|
assert [letter for letter, _ in gate.CLASSES] == ["a", "b", "c", "d", "e"]
|
|
assert all(description for _, description in gate.CLASSES)
|
|
|
|
|
|
# --- the eight attacks PM ran against this gate 2026-09-19 ---------------------
|
|
#
|
|
# Four of them went through: a row went GREEN without one label becoming true or
|
|
# one concept ranking better. Each is reproduced here as a test that must REFUSE
|
|
# it, and the four that were already refused stay as regression guards, so the
|
|
# table is 8 of 8 rather than 4 of 4.
|
|
|
|
|
|
def _hitting_set(set_id: str, *, quote: str, entries: int = 1) -> gate.QuestionSet:
|
|
"""A set the positive bundle answers, named by concept: the shortest way to
|
|
drive row 8 green without an adapter between the set and the row.
|
|
|
|
`entries` is load-bearing for the headline test: one question carrying TWO
|
|
fasit entries is the only shape in which "questions answered" and "units
|
|
hit" are different numbers, and a test where they agree cannot see a
|
|
headline that sums the units.
|
|
"""
|
|
fasit = [
|
|
gate.Fasit(by="concept", value="haandbok/vinterberedskap", quote=quote),
|
|
gate.Fasit(
|
|
by="concept",
|
|
value="haandbok/noekkelkvittering",
|
|
quote="kvitteres ut mot signatur" if quote else "",
|
|
),
|
|
][:entries]
|
|
return gate.QuestionSet(
|
|
set_id=set_id,
|
|
bundle="positive",
|
|
path=Path(f"/nowhere/{set_id}.json"),
|
|
sha256="0" * 64,
|
|
questions=(
|
|
gate.Question(
|
|
id=f"{set_id}-1",
|
|
question=(
|
|
"Naar kontrolleres vinterberedskapen paa hytta, og hvordan "
|
|
"kvitteres noekkelen ut?"
|
|
),
|
|
fasit=tuple(fasit),
|
|
),
|
|
),
|
|
controls=(),
|
|
)
|
|
|
|
|
|
def test_j2_row_eight_is_not_run_when_a_required_set_is_left_out(tmp_path: Path) -> None:
|
|
"""PM's J2: one set of three came back `6 of 6 GREEN`."""
|
|
bundles = {"positive": _bundles(tmp_path)["positive"]}
|
|
row = gate.row_eight([(_hitting_set("wiki-20", quote="innen 1. november"), bundles)])
|
|
assert row.status == gate.NOT_RUN
|
|
assert row.fails
|
|
assert "r761-sk2" in row.reason and "vegnormal-32" in row.reason
|
|
# The numbers it DID measure are still carried: a missing set must not cost
|
|
# the reader the set that ran.
|
|
assert any("wiki-20: " in detail for detail in row.details)
|
|
|
|
|
|
def test_row_eight_is_green_only_with_all_three_named_sets(tmp_path: Path) -> None:
|
|
bundles = {"positive": _bundles(tmp_path)["positive"]}
|
|
real = [
|
|
(_hitting_set(name, quote="innen 1. november"), bundles)
|
|
for name in ("wiki-20", "r761-sk2", "vegnormal-32")
|
|
]
|
|
row = gate.row_eight(real)
|
|
assert row.status == gate.GREEN
|
|
assert (row.k, row.m) == (3, 3)
|
|
|
|
|
|
def test_j2b_row_eight_never_sums_the_two_granularities_into_its_headline(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""PM's J2b: the headline was `quoted + concept` on the line above a detail
|
|
saying the two are not summed."""
|
|
bundles = {"positive": _bundles(tmp_path)["positive"]}
|
|
real = [
|
|
(_hitting_set("wiki-20", quote="innen 1. november", entries=2), bundles),
|
|
(_hitting_set("r761-sk2", quote=""), bundles),
|
|
(_hitting_set("vegnormal-32", quote=""), bundles),
|
|
]
|
|
row = gate.row_eight(real)
|
|
# Three questions, one per set: the headline is at QUESTION granularity.
|
|
# The sum it must NOT be is 4 of 4 -- two citation units plus two concept
|
|
# units -- which is why the first set carries two fasit entries.
|
|
assert (row.k, row.m) == (3, 3)
|
|
assert "question" in row.reason
|
|
assert any(
|
|
"2 of 2 at citation granularity, 2 of 2 at concept granularity" in detail
|
|
for detail in row.details
|
|
)
|
|
|
|
|
|
def _attack(tmp_path: Path, **cut: object) -> dict[int, gate.Row]:
|
|
"""Rows 1-6 remeasured under a cut this gate did not ship."""
|
|
cases, _ = gate.synthetic_cases(tmp_path / "bundles", FIXTURES)
|
|
with gate._wrap_cut(**cut):
|
|
remeasured = [gate.measure_case(case.question_set, case.bundles) for case in cases]
|
|
return {row.number: row for row in gate.deterministic_rows(remeasured)}
|
|
|
|
|
|
def test_j10_a_wider_cut_does_not_make_rows_two_and_three_green(tmp_path: Path) -> None:
|
|
"""PM's J10: `k = 32` took rows 1, 2, 3 and 6 green at once, and not one
|
|
label had become true -- the denominator of rows 2 and 3 IS the misses, so
|
|
delivering more broadly shrinks it to the cases that were already honest."""
|
|
baseline = _attack(tmp_path)
|
|
rows = _attack(tmp_path, k=32)
|
|
assert rows[2].status == gate.RED, rows[2]
|
|
assert rows[3].status != gate.GREEN, rows[3]
|
|
assert rows[3].fails
|
|
# The three fixtures that declare class b are delivered under this cut, so
|
|
# their premise no longer holds -- and a broken premise counts against the
|
|
# row rather than leaving it. NEITHER denominator shrinks: that is the whole
|
|
# of the defect.
|
|
assert any("premise" in detail for detail in rows[2].details), rows[2].details
|
|
assert rows[2].m >= baseline[2].m
|
|
assert rows[3].m >= baseline[3].m
|
|
|
|
|
|
def test_j8_removing_the_quota_leaves_row_three_unable_to_say_anything(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""PM's J8: with `--source-quota` off every printed reason became true and
|
|
row 3 read `6 of 6 GREEN`, while row 1 fell to 8 of 9. The label this row
|
|
judges was not printed at all, so the row did not measure."""
|
|
rows = _attack(tmp_path, source_quota=None)
|
|
assert rows[3].status == gate.NOT_RUN, rows[3]
|
|
assert "quota" in rows[3].reason
|
|
|
|
|
|
def test_a_forced_fixture_that_stops_missing_counts_against_row_two(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The fixed denominator, driven through a set file: a question that
|
|
DECLARES the class it forces and then comes back a hit has not been
|
|
classified -- its fixture's premise broke, and the row must carry it rather
|
|
than lose it."""
|
|
path, sha = _set_file(
|
|
tmp_path / "set.json",
|
|
set_id="row-two-premise",
|
|
bundle="positive",
|
|
questions=[
|
|
{
|
|
"id": "Z9",
|
|
"question": "Naar kontrolleres vinterberedskapen paa hytta?",
|
|
"expect_class": "b",
|
|
"fasit": [
|
|
{
|
|
"by": "concept",
|
|
"value": "haandbok/vinterberedskap",
|
|
"quote": "innen 1. november",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
)
|
|
case = gate.measure_case(gate.load_set(path, sha), _bundles(tmp_path))
|
|
assert [unit.hit for unit in case.units] == [True], "the fixture must HIT for this test"
|
|
row = gate.row_two([case])
|
|
assert (row.k, row.m, row.status) == (0, 1, gate.RED)
|
|
assert any("Z9" in detail and "premise" in detail for detail in row.details)
|
|
|
|
|
|
def test_j1_a_registration_this_session_wrote_is_not_a_hold_out(tmp_path: Path) -> None:
|
|
"""PM's J1: two files written by the session under test came back
|
|
`7 of 7 GREEN`. Every check was an assertion the registration made about
|
|
itself -- `written_by` is `bool()` of a string the file sets, and the
|
|
readings are read from a list the same file owns."""
|
|
registration = _registration(
|
|
tmp_path,
|
|
set_path=FIXTURES / "set-positive.json",
|
|
bundle=_bundles(tmp_path)["positive"],
|
|
threshold=0.8,
|
|
)
|
|
spec = json.loads(registration.read_text(encoding="utf-8"))
|
|
spec["written_by"] = "the same session, lying"
|
|
registration.write_text(json.dumps(spec), encoding="utf-8")
|
|
row = gate.row_five(registration)
|
|
assert row.status != gate.GREEN, row
|
|
assert row.fails
|
|
# What a session cannot write about itself: that git already carried the
|
|
# file before the ranking moved.
|
|
assert any("committed" in detail for detail in row.details), row.details
|
|
assert any("ranking" in detail for detail in row.details), row.details
|
|
|
|
|
|
def test_j3_row_seven_refuses_a_mutant_list_that_is_not_the_pinned_roster(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""PM's J3: seven duplicate `k = 1` mutants appended to `MUTANTS` took the
|
|
row from 11 of 13 RED to 18 of 20 GREEN with the SAME two survivors. The
|
|
bar was 90 % of a list living in the file a capability session edits."""
|
|
cases, _ = gate.synthetic_cases(tmp_path / "bundles", FIXTURES)
|
|
baseline = gate.deterministic_rows(cases)
|
|
padded = (*gate.MUTANTS, *(gate.MUTANTS[2] for _ in range(7)))
|
|
row = gate.row_seven(cases, baseline, mutants=padded)
|
|
assert row.status != gate.GREEN, row
|
|
assert row.fails
|
|
assert "roster" in row.reason or "duplicate" in row.reason
|
|
|
|
|
|
def test_g9_a_delivery_that_carries_the_citation_and_not_the_bundles_bytes_is_not_a_hit(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""PM's G9: `hit = bool(hit_ids) and bool(confirmed)` -- removing the
|
|
second term left all 46 tests green, because every mutation that touched
|
|
the text emptied `hit_ids` one step earlier. The shape that reaches the
|
|
term is a delivery that still CARRIES the quote and is no longer the
|
|
concept file's bytes, and the judge must have read the bundle first."""
|
|
bundles = _bundles(tmp_path)
|
|
gate.bundle_index(bundles["positive"]) # the judge reads the bundle BEFORE the payload
|
|
question_set = gate.load_set(
|
|
FIXTURES / "set-positive.json", gate.SYNTHETIC_SETS["set-positive.json"]
|
|
)
|
|
with gate._extend_delivered():
|
|
units = gate.measure_case(question_set, bundles).units
|
|
assert units, "the known-positive: this set delivers on the unmutated run"
|
|
assert [unit.hit for unit in units] == [False] * len(units)
|
|
assert all(unit.confirmed is False for unit in units)
|
|
assert all(unit.detail == "the delivered text is not the bundle's bytes" for unit in units)
|
|
|
|
|
|
def test_the_unmutated_run_of_that_same_set_is_every_hit(tmp_path: Path) -> None:
|
|
units = _case(tmp_path, "set-positive.json").units
|
|
assert [unit.hit for unit in units] == [True] * len(units)
|
|
|
|
|
|
def test_the_mutant_roster_is_pinned_apart_from_the_list_it_names() -> None:
|
|
assert tuple(mutant.label for mutant in gate.MUTANTS) == gate.MUTANT_ROSTER
|
|
assert len(gate.MUTANT_ROSTER) == gate.MUTANT_COUNT
|
|
assert len(set(gate.MUTANT_ROSTER)) == len(gate.MUTANT_ROSTER)
|
|
|
|
|
|
def test_the_synthetic_corpus_is_pinned_like_the_sets(tmp_path: Path) -> None:
|
|
assert gate.specs_digest() == gate.SPECS_SHA256
|
|
tuned = dict(gate.SPECS)
|
|
first = tuned["positive"]
|
|
tuned["positive"] = gate.BundleSpec(first.bundle_id, first.documents[:1])
|
|
assert gate.specs_digest(tuned) != gate.SPECS_SHA256
|
|
|
|
|
|
def test_a_corpus_that_is_not_the_pinned_corpus_is_refused(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(gate, "SPECS_SHA256", "0" * 64)
|
|
with pytest.raises(gate.GateUsage) as refusal:
|
|
gate.synthetic_bundles(tmp_path / "bundles")
|
|
assert "not the corpus that was pinned" in str(refusal.value)
|
|
|
|
|
|
def _k2_shaped(tmp_path: Path, questions: int) -> tuple[Path, str]:
|
|
"""A K2 gold set in this gate's own shape. `set-positive.json` carries
|
|
exactly six questions the bundle answers, which is K2's denominator."""
|
|
spec = json.loads((FIXTURES / "set-positive.json").read_text(encoding="utf-8"))
|
|
spec["set_id"] = "k2-gold"
|
|
spec["questions"] = spec["questions"][:questions]
|
|
path = tmp_path / "k2.json"
|
|
path.write_text(json.dumps(spec, ensure_ascii=False), encoding="utf-8")
|
|
return path, gate.sha256_of(path)
|
|
|
|
|
|
def test_row_nine_is_green_when_a_k2_gold_set_arrives(tmp_path: Path) -> None:
|
|
path, sha = _k2_shaped(tmp_path, gate.K2_QUESTIONS)
|
|
question_set = gate.load_set(path, sha)
|
|
row = gate.row_nine((question_set, _bundles(tmp_path)))
|
|
assert (row.k, row.m, row.status) == (6, 6, gate.GREEN)
|
|
|
|
|
|
def test_row_nine_is_red_when_the_gold_set_is_not_answered(tmp_path: Path) -> None:
|
|
path, sha = _k2_shaped(tmp_path, gate.K2_QUESTIONS)
|
|
question_set = gate.load_set(path, sha)
|
|
bundles = dict(_bundles(tmp_path))
|
|
bundles["positive"] = bundles["miss"] # the same six questions, the wrong bundle
|
|
row = gate.row_nine((question_set, bundles))
|
|
assert (row.k, row.m, row.status) == (0, 6, gate.RED)
|
|
|
|
|
|
def test_a_k2_set_of_another_size_is_another_set_and_is_refused(tmp_path: Path) -> None:
|
|
path, sha = _k2_shaped(tmp_path, gate.K2_QUESTIONS - 1)
|
|
with pytest.raises(gate.GateUsage) as refusal:
|
|
gate._k2_set([str(path), sha, str(tmp_path)])
|
|
assert "K2's denominator" in str(refusal.value)
|
|
|
|
|
|
def test_j7_a_cut_that_delivers_nothing_is_exit_two_and_never_a_quiet_green_row(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""PM's J7/J11, kept as a regression guard: a cut delivering nothing came
|
|
back as an error from `consume` and the gate exited 2. A row counting
|
|
nothing must never read as a row that found nothing wrong."""
|
|
cases, _ = gate.synthetic_cases(tmp_path / "bundles", FIXTURES)
|
|
with gate._wrap_cut(k=0), pytest.raises(Exception) as refusal:
|
|
[gate.measure_case(case.question_set, case.bundles) for case in cases]
|
|
assert not isinstance(refusal.value, AssertionError), refusal.value
|
|
|
|
|
|
# --- step 0: the real sets are pinned, and row 8 names what it measured -------
|
|
|
|
|
|
def test_row_eight_names_the_bundle_identity_of_every_set_it_measured(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The row printed the SET's sha256 and never the bundle's identity.
|
|
|
|
Measured 2026-09-19 by PM with one pinned r761 set: three bundles gave
|
|
4 of 7 and a fourth gave 0 of 7, while the session recorded 7 of 7 -- from
|
|
a bundle no line of the output named. A number nobody can attach to a
|
|
bundle cannot be reproduced and cannot be felled.
|
|
"""
|
|
bundles = {"positive": _bundles(tmp_path)["positive"]}
|
|
real = [
|
|
(_hitting_set(name, quote="innen 1. november"), bundles) for name in gate.REQUIRED_REAL_SETS
|
|
]
|
|
row = gate.row_eight(real)
|
|
# The identity is read here, independently of the gate, from the same
|
|
# bundle the row was handed.
|
|
bundle_id = consume.root_bundle_id_of(bundles["positive"])
|
|
ref = consume.bundle_ref(bundles["positive"])
|
|
assert bundle_id and ref.startswith("sha256-tree:")
|
|
for set_id in gate.REQUIRED_REAL_SETS:
|
|
named = [detail for detail in row.details if detail.strip().startswith(f"{set_id}:")]
|
|
assert named, f"{set_id} has no line of its own"
|
|
block = "\n".join(row.details)
|
|
assert bundle_id in block, f"{set_id}: the bundle_id is nowhere in the output"
|
|
assert ref[:24] in block, f"{set_id}: the bundle ref is nowhere in the output"
|
|
|
|
|
|
def test_all_three_real_sets_are_pinned_and_not_only_the_first(tmp_path: Path) -> None:
|
|
"""Every one of the three, driven through the command-line path.
|
|
|
|
The names and the count are written out here rather than read from
|
|
`REQUIRED_REAL_SETS`: a test taking its denominator from the tuple it is
|
|
checking would stay green if a name were dropped from both.
|
|
"""
|
|
names = [("wiki", "wiki-20"), ("r761", "r761-sk2"), ("vegnormal", "vegnormal-32")]
|
|
assert len(names) == 3
|
|
bundle = _bundles(tmp_path)["positive"]
|
|
for adapter, set_id in names:
|
|
path = tmp_path / f"{adapter}-invented.json"
|
|
path.write_text(_invented_real_set(adapter), encoding="utf-8")
|
|
with pytest.raises(gate.GateUsage) as refusal:
|
|
gate._real_sets([(adapter, str(path), gate.sha256_of(path), str(bundle))])
|
|
assert set_id in str(refusal.value)
|
|
|
|
|
|
def _invented_real_set(adapter: str) -> str:
|
|
"""One question, in the adapter's own shape, with invented content."""
|
|
if adapter == "wiki":
|
|
return json.dumps(
|
|
{
|
|
"questions": [
|
|
{
|
|
"id": "W1",
|
|
"question": "Naar kontrolleres vinterberedskapen?",
|
|
"fasit": [{"doc": "haandbok", "quote": "innen 1. november"}],
|
|
}
|
|
]
|
|
}
|
|
)
|
|
if adapter == "r761":
|
|
return json.dumps(
|
|
{"sporsmal": [{"id": "S1", "sporsmal": "Hva sier punktet?", "fasit": "4.2 Vakthold"}]}
|
|
)
|
|
return json.dumps(
|
|
{
|
|
"sporsmal": [
|
|
{
|
|
"id": "T1-1",
|
|
"sporsmal": "Hva viser kravet til?",
|
|
"must_cite": [{"normal": "N100:2023", "req_number": "2.3.2-3"}],
|
|
}
|
|
]
|
|
}
|
|
)
|
|
|
|
|
|
def test_a_wiki_set_of_one_question_is_refused_on_the_command_line(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""PM's measured attack: three self-written one-question files and a
|
|
self-written bundle read `3 of 3 GREEN`, because both the set AND its
|
|
sha256 came from the command line and nothing said how big `wiki-20` is.
|
|
"""
|
|
path = tmp_path / "wiki-shaped.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema": "1",
|
|
"questions": [
|
|
{
|
|
"id": "W1",
|
|
"question": "Naar kontrolleres vinterberedskapen?",
|
|
"fasit": [{"doc": "haandbok", "quote": "innen 1. november"}],
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
bundle = _bundles(tmp_path)["positive"]
|
|
with pytest.raises(gate.GateUsage) as refusal:
|
|
gate._real_sets([("wiki", str(path), gate.sha256_of(path), str(bundle))])
|
|
assert "wiki-20" in str(refusal.value)
|
|
|
|
|
|
# --- step 0: the threshold is a number, and it is compared with something ----
|
|
|
|
|
|
def _registration(
|
|
tmp_path: Path,
|
|
*,
|
|
set_path: Path,
|
|
bundle: Path,
|
|
threshold: object,
|
|
name: str = "registration.json",
|
|
) -> Path:
|
|
registration = tmp_path / name
|
|
registration.write_text(
|
|
json.dumps(
|
|
{
|
|
"set": str(set_path),
|
|
"sha256": gate.sha256_of(set_path),
|
|
"bundle": str(bundle),
|
|
"metric": "questions answered over questions asked",
|
|
"threshold": threshold,
|
|
"threshold_written_at": "2026-09-19T10:00:00Z",
|
|
"written_by": "the session that wrote the eval",
|
|
"readings": [{"at": "2026-09-20T09:00:00Z", "value": "not read yet"}],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return registration
|
|
|
|
|
|
def test_a_threshold_that_is_not_a_number_is_refused(tmp_path: Path) -> None:
|
|
"""`bool(threshold)` was the whole check, so `report-only; any number is
|
|
acceptable for v1` read as `a threshold is written: yes`. A threshold that
|
|
cannot be compared with a number cannot fell anything."""
|
|
bundles = _bundles(tmp_path)
|
|
registration = _registration(
|
|
tmp_path,
|
|
set_path=FIXTURES / "set-positive.json",
|
|
bundle=bundles["positive"],
|
|
threshold="report-only; any number is acceptable for v1",
|
|
)
|
|
row = gate.row_five(registration, provenance=lambda _: _carried_by_git())
|
|
assert any("the threshold is a number: NO" in detail for detail in row.details)
|
|
assert row.fails
|
|
|
|
|
|
def test_the_threshold_is_compared_with_the_measured_hold_out(tmp_path: Path) -> None:
|
|
"""Both directions, from the same code path: a set the bundle answers
|
|
clears a threshold under it, and a set it does not answer falls under one
|
|
over it. The measured share is counted here as well, off the set's own
|
|
questions, so the row is not the only thing that knows it."""
|
|
bundles = _bundles(tmp_path)
|
|
clears = gate.row_five(
|
|
_registration(
|
|
tmp_path,
|
|
set_path=FIXTURES / "set-positive.json",
|
|
bundle=bundles["positive"],
|
|
threshold=0.8,
|
|
name="clears.json",
|
|
),
|
|
provenance=lambda _: _carried_by_git(),
|
|
)
|
|
assert any("clears the threshold: yes" in detail for detail in clears.details)
|
|
assert (clears.k, clears.m, clears.status) == (11, 11, gate.GREEN)
|
|
|
|
# The independent count: set-miss carries one question, forced to miss.
|
|
missing = json.loads((FIXTURES / "set-miss.json").read_text(encoding="utf-8"))
|
|
assert len(missing["questions"]) == 1
|
|
falls = gate.row_five(
|
|
_registration(
|
|
tmp_path,
|
|
set_path=FIXTURES / "set-miss.json",
|
|
bundle=bundles["miss"],
|
|
threshold=0.5,
|
|
name="falls.json",
|
|
),
|
|
provenance=lambda _: _carried_by_git(),
|
|
)
|
|
assert any("clears the threshold: NO" in detail for detail in falls.details)
|
|
assert (falls.k, falls.m, falls.status) == (10, 11, gate.RED)
|
|
|
|
|
|
# --- step 1: row 4, the marking a consumer can act on -------------------------
|
|
|
|
|
|
def test_row_four_marks_every_uncovered_control_and_leaves_the_covered_one(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Six controls, five uncovered and one covered, and the row must get all
|
|
six right.
|
|
|
|
Three of the five were measured red on 2026-09-19: N3, N4 and N5 came back
|
|
with 8, 8 and 1 excerpts and nothing in the payload that said they were
|
|
weak, so `marked = nothing delivered` could not see them.
|
|
"""
|
|
case = _case(tmp_path, "set-controls.json")
|
|
# The denominator, counted from the pinned file rather than from the run.
|
|
declared = json.loads((FIXTURES / "set-controls.json").read_text(encoding="utf-8"))
|
|
assert len(declared["controls"]) == 6
|
|
assert sum(1 for control in declared["controls"] if control.get("covered")) == 1
|
|
row = gate.row_four([case])
|
|
assert (row.k, row.m, row.status) == (6, 6, gate.GREEN)
|
|
assert not row.details
|
|
|
|
|
|
def test_the_marking_stays_off_every_question_the_synthetic_sets_do_answer(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The known-negative, and it is wider than row 4's own denominator.
|
|
|
|
A marking that fires on the questions the bundles DO answer would take row
|
|
4 to 6 of 6 and be worthless. Every question of the three sets whose fasit
|
|
row 1 finds is measured here, and none of them may come back marked.
|
|
"""
|
|
bundles = _bundles(tmp_path)
|
|
answered = 0
|
|
for name in ("set-positive.json", "set-signals.json", "set-quota.json"):
|
|
question_set = gate.load_set(FIXTURES / name, gate.SYNTHETIC_SETS[name])
|
|
for question in question_set.questions:
|
|
bundle = bundles[question.bundle or question_set.bundle]
|
|
payload = consume.build_payload(
|
|
bundle, question=question.question, k=question.k, limit=question.limit
|
|
)
|
|
assert not gate.marked(payload), f"{question_set.set_id}/{question.id}"
|
|
answered += 1
|
|
assert answered == 10, answered
|
|
|
|
|
|
def test_a_set_of_the_right_size_and_the_wrong_bytes_is_still_refused(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The size pin alone would be a one-line forgery: twenty invented
|
|
questions in the wiki shape carry the pinned count and none of the pinned
|
|
content."""
|
|
path = tmp_path / "wiki-sized.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"questions": [
|
|
{
|
|
"id": f"W{index}",
|
|
"question": "Naar kontrolleres vinterberedskapen?",
|
|
"fasit": [
|
|
{"doc": "haandbok", "quote": "innen 1. november"},
|
|
*([{"doc": "haandbok", "quote": ""}] if index < 9 else []),
|
|
],
|
|
}
|
|
for index in range(20)
|
|
]
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
# The counts this file carries ARE the pinned ones, measured here.
|
|
question_set = gate.read_real_set("wiki", path, gate.sha256_of(path))
|
|
pinned = gate.REAL_SET_PINS["wiki-20"]
|
|
assert (len(question_set.questions), question_set.units) == (
|
|
pinned.questions,
|
|
pinned.fasit_entries,
|
|
)
|
|
with pytest.raises(gate.GateUsage) as refusal:
|
|
gate._real_sets(
|
|
[("wiki", str(path), gate.sha256_of(path), str(_bundles(tmp_path)["positive"]))]
|
|
)
|
|
assert "sha256" in str(refusal.value)
|
|
|
|
|
|
def test_a_threshold_outside_nought_to_one_is_not_a_share(tmp_path: Path) -> None:
|
|
"""`80` is either 80 % written wrongly or a bar no run can clear, and
|
|
guessing which is not this row's job."""
|
|
bundles = _bundles(tmp_path)
|
|
for value in (80, -0.1, 1.5):
|
|
row = gate.row_five(
|
|
_registration(
|
|
tmp_path,
|
|
set_path=FIXTURES / "set-positive.json",
|
|
bundle=bundles["positive"],
|
|
threshold=value,
|
|
name=f"t{value}.json",
|
|
),
|
|
provenance=lambda _: _carried_by_git(),
|
|
)
|
|
assert any("the threshold is a number: NO" in detail for detail in row.details), value
|
|
# The control: a share inside the range is accepted as one.
|
|
inside = gate.row_five(
|
|
_registration(
|
|
tmp_path,
|
|
set_path=FIXTURES / "set-positive.json",
|
|
bundle=bundles["positive"],
|
|
threshold=1.0,
|
|
name="inside.json",
|
|
),
|
|
provenance=lambda _: _carried_by_git(),
|
|
)
|
|
assert any("the threshold is a number: yes" in detail for detail in inside.details)
|
|
|
|
|
|
def test_a_hold_out_set_with_no_question_clears_no_threshold(tmp_path: Path) -> None:
|
|
"""A share over a denominator of nought is not a number, and an empty set
|
|
was the shape every row-5 test used before 2026-09-20 -- so `>=` over it
|
|
would have made the comparison vacuous the moment it was added."""
|
|
empty = tmp_path / "empty-set.json"
|
|
empty.write_text(
|
|
json.dumps({"set_id": "empty", "bundle": "positive", "questions": []}),
|
|
encoding="utf-8",
|
|
)
|
|
row = gate.row_five(
|
|
_registration(
|
|
tmp_path,
|
|
set_path=empty,
|
|
bundle=_bundles(tmp_path)["positive"],
|
|
threshold=0.0,
|
|
),
|
|
provenance=lambda _: _carried_by_git(),
|
|
)
|
|
assert any(
|
|
"clears the threshold: NO (the hold-out set carries no question" in detail
|
|
for detail in row.details
|
|
), row.details
|