Kjell Tore Guttormsen
3d496de2c6
PM's checkpoint on 2c8296b ran eight cheating attacks at this gate and four
went through: a row came back GREEN without one label becoming true or one
concept ranking better. This is the first of them, written as a test that must
refuse it.
J2: `row_eight` iterates over whatever `--real` handed it and counts. One set
of three came back `6 of 6 GREEN`, and the realistic route is `--real r761`
alone -- the set PM's own registration puts at 7 of 7, with the two that miss
left out. The docstring already said "never green by leaving a set out"; the
code did not, and THIS REPOSITORY'S OWN TEST asserted the opposite at
`tests/test_retrieval_gate.py:486`, `(1, 1, GREEN)` for a single wiki set.
That assertion is corrected here rather than worked around: a test that pins
the defect is the defect.
J2b: the headline `k of N` is `quoted_hits + concept_hits` over
`quoted_units + concept_units`, written on the line directly above the detail
that says the two granularities are NOT summed into one number.
3 red on an assertion about behaviour (GREEN where NOT RUN is required, and a
headline that is a sum), 46 passed. The green direction -- all three named
sets present -- is in the same commit and passes already, so the fix cannot
close the rows by making the row unreachable.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
816 lines
31 KiB
Python
816 lines
31 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_is_red_because_the_payload_names_the_quota_for_a_rank(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
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)
|
|
|
|
|
|
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_is_red_because_an_uncovered_question_still_gets_excerpts(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
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 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_seven(tmp_path: Path) -> None:
|
|
held = tmp_path / "held.json"
|
|
held.write_text('{"questions": []}', encoding="utf-8")
|
|
registration = tmp_path / "registration.json"
|
|
registration.write_text(
|
|
json.dumps(
|
|
{
|
|
"set": str(held),
|
|
"sha256": gate.sha256_of(held),
|
|
"threshold": "hit@payload at unit granularity >= 0.8",
|
|
"threshold_written_at": "2026-09-19T10:00:00Z",
|
|
"written_by": "the session that wrote the eval, never the one that "
|
|
"changes the ranking",
|
|
"readings": [{"at": "2026-09-20T09:00:00Z", "value": "not read yet"}],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
row = gate.row_five(registration)
|
|
assert (row.k, row.m, row.status) == (7, 7, gate.GREEN)
|
|
|
|
|
|
def test_row_five_falls_on_a_number_read_before_its_threshold_was_written(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
held = tmp_path / "held.json"
|
|
held.write_text('{"questions": []}', encoding="utf-8")
|
|
registration = tmp_path / "registration.json"
|
|
registration.write_text(
|
|
json.dumps(
|
|
{
|
|
"set": str(held),
|
|
"sha256": gate.sha256_of(held),
|
|
"threshold": "hit@payload at unit granularity >= 0.8",
|
|
"threshold_written_at": "2026-09-19T10:00:00Z",
|
|
"written_by": "somebody",
|
|
"readings": [{"at": "2026-09-18T09:00:00Z", "value": "0.62"}],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
row = gate.row_five(registration)
|
|
assert (row.k, row.m, row.status) == (6, 7, gate.RED)
|
|
assert any("no reading predates the threshold: NO" in detail for detail in 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)
|
|
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)
|
|
row = gate.row_seven(cases, baseline, mutants=(_noop_mutant(),))
|
|
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_that_only_makes_a_red_row_greener_is_not_felled(tmp_path: Path) -> None:
|
|
# The definition, held by a test: removing the quota makes row 3's label
|
|
# honest, because there is then no quota to name. Calling that a kill
|
|
# would credit this gate with a check it does not have.
|
|
cases, _ = gate.synthetic_cases(tmp_path / "bundles", FIXTURES)
|
|
baseline = gate.deterministic_rows(cases)
|
|
before = {row.number: (row.k, row.m) for row in baseline}
|
|
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, row.m) for row in rows}
|
|
assert after[3][0] - after[3][1] > before[3][0] - before[3][1]
|
|
|
|
|
|
# --- 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] == [3, 4, 5, 7, 8, 9]
|
|
assert (by_number[1].k, by_number[1].m) == (9, 9)
|
|
assert (by_number[2].k, by_number[2].m) == (7, 7)
|
|
assert (by_number[6].k, by_number[6].m) == (9, 9)
|
|
|
|
|
|
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 3, 4, 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) -> 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."""
|
|
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?",
|
|
fasit=(
|
|
gate.Fasit(
|
|
by="concept", value="haandbok/vinterberedskap", quote=quote
|
|
),
|
|
),
|
|
),
|
|
),
|
|
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"), 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 and
|
|
# is never the sum of one citation unit and two concept units.
|
|
assert (row.k, row.m) == (3, 3)
|
|
assert "question" in row.reason
|
|
assert any(
|
|
"1 of 1 at citation granularity, 2 of 2 at concept granularity" in detail
|
|
for detail in row.details
|
|
)
|