open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 4 Activity
llm-ingestion-okf/tests/test_retrieval_gate.py
Kjell Tore Guttormsen 70c388d419 test(retrieval-gate): row 8's judge must read a passage as a passage — red 
Since v1.1 C1 a concept over PASSAGE_CHARS is delivered as its answering
passage (heading, [...], span, [...]), and the judge still demands the whole
body (the row-6 identity), so every such delivery is class e even when the span
carries the citation. PM's re-measurement of e503f6a found this to be the
dominant miss class of row 8 on the real set.

Pinned before the judge moves, on an invented one-concept bundle:
- a passage whose span carries the quote is a HIT; the offsets count in the
  DELIVERED body, and a test asserts that reading them in the concept file
  misses (trap 1);
- the same passage with one invented sentence, in the span or between the
  marker line and the span, is class e, named (trap 2: "the span occurs
  somewhere in the text" would count it);
- a quote the passage carries only in its heading or its markers is class d,
  beside the known-positive in the same run;
- row 7 carries M15, passage cheating, as its fifteenth mutant.

Fails on e503f6a: five assertion failures and one ValueError (M15 does not
exist yet); no collection error.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-22 23:53:05 +02:00

1509 lines
60 KiB
Python
Raw Blame History

"""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 set in
`claude-code-llm-wiki` is never read by a test: its adapter is exercised
against files written in its shape, with invented content, inside
`tmp_path`. The test track built on material tied to the operator's
employer was retired 2026-09-21, and with it the gate's row 9 and two
adapters; nothing below reads, needs or names that material.
"""
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 bm25, 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",
# Shares no word with the fasit concept, so no ranking can
# deliver it -- the row's red state is forced by the fixture.
"question": "Hvem eier kanoen ved brygga?",
"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: [])
# And its second reading since v1.1 C4: no word is absent in any form.
monkeypatch.setattr(bm25, "_absent", 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]
@pytest.mark.parametrize(
"prefix",
[
# The three that survived v1.1: each is a mechanism the default ranking
# RUNS, so a gate that cannot see it removed cannot see it break.
"M06 ", # the passage signal reads no body
"M07 ", # the field signal weighs no title and no path
"M10 ", # the fusion is flattened
# A passage carrying one sentence the file does not: M14's shape for
# the delivery form v1.1 C1 added.
"M15 ",
],
)
def test_row_seven_fells_every_mutant_of_a_mechanism_the_ranking_runs(
tmp_path: Path, prefix: str
) -> None:
"""A survivor of row 7 is a mechanism the synthetic corpus never makes
decide a delivery. Felled means a row got WORSE on the pinned sets and the
pinned corpus -- never a row that merely moved."""
(mutant,) = [m for m in gate.MUTANTS if m.label.startswith(prefix)]
cases, _ = gate.synthetic_cases(tmp_path / "bundles", FIXTURES)
before = {row.number: row.k - row.m for row in gate.deterministic_rows(cases)}
with mutant.patch():
rows = gate.deterministic_rows(
[gate.measure_case(case.question_set, case.bundles) for case in cases]
)
after = {row.number: row.k - row.m for row in rows}
worse = [number for number in sorted(before) if after[number] < before[number]]
assert worse, f"{mutant.label} survives: no row got worse ({before} -> {after})"
# --- row 8 --------------------------------------------------------------------
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"]})])
# The one required set, answered: row 8's own green direction. What
# refuses a self-written file of this shape is the pin on the command-line
# path (`test_a_wiki_set_of_one_question_is_refused_on_the_command_line`).
assert (row.k, row.m, row.status) == (1, 1, gate.GREEN)
assert any("citation granularity" in detail for detail in row.details)
def test_an_unknown_real_set_name_is_refused(tmp_path: Path) -> None:
path = tmp_path / "x.json"
path.write_text("{}", encoding="utf-8")
# The two retired adapters are unknown names now, like any other.
for name in ("something-else", "r761", "vegnormal"):
with pytest.raises(gate.GateUsage) as refusal:
gate.read_real_set(name, path, gate.sha256_of(path))
assert "unknown real set" in str(refusal.value)
# --- 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]
# Rows 2 and 3 are green again since the synthetic corpus was re-measured
# for BM25 (2026-09-21). Row 7 fells all 14 since 2026-09-22, when the three
# v1.1 survivors each got a fixture that makes their mechanism decide a
# delivery (`set-mechanisms.json`, three more row-1 units).
assert [row.number for row in rows if row.fails] == [5, 8]
assert (by_number[1].k, by_number[1].m) == (13, 13)
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) == (13, 13)
assert (by_number[7].k, by_number[7].m) == (14, 14)
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, 9):
assert f"\n{number} " in f"\n{printed}"
assert "GATE RED: rows 5, 8" 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, 9))
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`. The rule outlived
the two retired sets: a set that is not the required one measures
something, and the row is still not a measurement of the required set."""
bundles = {"positive": _bundles(tmp_path)["positive"]}
row = gate.row_eight([(_hitting_set("some-other-set", quote="innen 1. november"), bundles)])
assert row.status == gate.NOT_RUN
assert row.fails
assert "wiki-20" 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("some-other-set: " in detail for detail in row.details)
def test_row_eight_is_green_only_with_every_named_set(tmp_path: Path) -> None:
bundles = {"positive": _bundles(tmp_path)["positive"]}
real = [(_hitting_set(name, quote="innen 1. november"), bundles) for name in ("wiki-20",)]
row = gate.row_eight(real)
assert row.status == gate.GREEN
assert (row.k, row.m) == (1, 1)
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("an-invented-concept-set", quote=""), bundles),
]
row = gate.row_eight(real)
# Two questions, one per set: the headline is at QUESTION granularity.
# The sum it must NOT be is 3 of 3 -- two citation units plus one concept
# unit -- which is why the first set carries two fasit entries.
assert (row.k, row.m) == (2, 2)
assert "question" in row.reason
assert any(
"2 of 2 at citation granularity, 1 of 1 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)
# --- a passage is judged as a passage ------------------------------------------
#
# Since v1.1 C1 a concept over `consume.PASSAGE_CHARS` is delivered as its
# answering passage: the nearest heading above the span, `[...]`, the span,
# `[...]`. The row-6 identity (delivered text == the whole body) is older than
# that and read every such delivery as class e. The judge must now accept a
# passage that IS its exact reconstruction from the bundle's bytes, and nothing
# looser: "the span occurs somewhere in the text" accepts an invented sentence
# beside it.
_HOUSE_NOTE = "Styret gjennomgaar notatet og foerer det inn i arkivet. Sekretaeren sender kopi."
_INVENTED = "En setning som ikke staar i konseptfila."
def _house_bundle(tmp_path: Path) -> Path:
"""One concept of about 6 600 characters: a heading ten lines in, the
answer thirty lines below it, and a literal `[...]` on the LAST line, far
outside any span centred on the answer."""
body = "\n".join(
[_HOUSE_NOTE] * 10
+ ["## Teknisk rom"]
+ [_HOUSE_NOTE] * 30
+ ["Varmepumpa i kjelleren faar service av roerleggeren hvert aar."]
+ [_HOUSE_NOTE] * 40
+ ["Se vedlegg [...] i arkivet."]
)
spec = gate.BundleSpec(
"retrieval-house",
(
gate.DocumentSpec(
"husbok", "husbok.md", (gate.ConceptSpec("drift", "Drift av huset", body),)
),
*(
gate.DocumentSpec(
f"skriv-{number}",
f"skriv-{number}.md",
(gate.ConceptSpec("skriv", f"Skriv {number}", "Skrivet er arkivert."),),
)
for number in range(3)
),
),
)
return gate.build_bundle(tmp_path / "house", spec)
def _house_question(*quotes: str) -> gate.Question:
return gate.Question(
id="H1",
question="Hvem gir varmepumpa i kjelleren service?",
fasit=tuple(gate.Fasit("concept", "husbok/drift", quote) for quote in quotes),
)
def _inject(where: str) -> pytest.MonkeyPatch:
"""A passage carrying one sentence the concept file does not: at the end of
the span, or between the heading's marker line and the span."""
original = consume.as_passage
def mutant(excerpt: dict[str, object], window: int) -> dict[str, object]:
out = original(excerpt, window)
if "passage" in out:
text = str(out["text"])
if where == "span":
cut = text.rindex(f"\n{consume.PASSAGE_ELISION}")
out["text"] = f"{text[:cut]} {_INVENTED}{text[cut:]}"
else:
marker = f"{consume.PASSAGE_ELISION}\n"
cut = text.index(marker) + len(marker)
out["text"] = f"{text[:cut]}{_INVENTED}\n{text[cut:]}"
return out
patch = pytest.MonkeyPatch()
patch.setattr(consume, "as_passage", mutant)
return patch
def test_a_passage_whose_span_carries_the_quote_is_a_hit(tmp_path: Path) -> None:
bundle = _house_bundle(tmp_path)
gate.bundle_index(bundle)
question = _house_question("service av roerleggeren")
payload = consume.build_payload(bundle, question=question.question, withheld_full=True)
(excerpt,) = [e for e in payload["excerpts"] if e["concept_id"] == "husbok/drift"]
passage = excerpt["passage"]
assert isinstance(passage, dict), "the premise: this concept is delivered as a passage"
start, end = passage["start"], passage["end"]
assert 0 < start and end < passage["of"], "the premise: `[...]` on both sides"
# FELLE 1, pinned: the offsets count in the DELIVERED body. Read against
# the concept file they land a frontmatter's length off.
view = gate.bundle_index(bundle).concepts["husbok/drift"]
assert view.body[start:end] in str(excerpt["text"])
assert view.whole[start:end] not in str(excerpt["text"])
(unit,) = gate.measure_units(bundle, question)
assert (unit.hit, unit.confirmed, unit.klass) == (True, True, None), unit
@pytest.mark.parametrize("where", ["span", "between heading and span"])
def test_a_passage_carrying_an_invented_sentence_is_not_a_hit(tmp_path: Path, where: str) -> None:
"""FELLE 2, pinned: the quote is still in the span and the span is still in
the text, so "the span occurs somewhere" would count this. The delivered
text must BE the reconstruction."""
bundle = _house_bundle(tmp_path)
gate.bundle_index(bundle) # the judge reads the bundle BEFORE the payload
question = _house_question("service av roerleggeren")
patch = _inject(where)
try:
(unit,) = gate.measure_units(bundle, question)
finally:
patch.undo()
assert unit.hit is False
assert unit.confirmed is False
assert unit.klass == "e"
assert unit.detail == "the delivered passage is not its reconstruction from the bundle's bytes"
def test_a_quote_the_passage_carries_only_outside_its_span_is_not_a_hit(tmp_path: Path) -> None:
"""The heading and the markers are delivered and are not the span: a quote
found only there was delivered without the citation (class d). The first
unit is the known-positive of the same run."""
bundle = _house_bundle(tmp_path)
units = gate.measure_units(
bundle, _house_question("service av roerleggeren", "Teknisk rom", "[...]")
)
assert [(unit.hit, unit.klass) for unit in units] == [
(True, None),
(False, "d"),
(False, "d"),
], units
def test_row_seven_names_passage_cheating() -> None:
assert gate.MUTANT_COUNT == 15
assert gate.MUTANT_ROSTER[-1] == "M15 the passage carries a sentence the concept file does not"
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 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 real 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_every_required_real_set_is_pinned(tmp_path: Path) -> None:
"""Every required set, 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")]
assert len(names) == len(gate.REQUIRED_REAL_SETS) == 1
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."""
assert adapter == "wiki", adapter
return json.dumps(
{
"questions": [
{
"id": "W1",
"question": "Naar kontrolleres vinterberedskapen?",
"fasit": [{"doc": "haandbok", "quote": "innen 1. november"}],
}
]
}
)
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
Reference in a new issue View git blame Copy permalink