The registered hold-out set is written in a consumer's own question-set
schema, the one row 8 already reads, and row 5 reads every set with the
gate's synthetic reader: "not a question set this gate can read:
'set_id'", a NO no set in that schema can turn.
Three tests over an invented set in that schema and the synthetic
bundle: 4 of 5 clears 0.8 and falls under 0.9 through the same code
path; an unknown schema is a NO that names the schema; and the row
prints the share and never which hold-out question missed, with the
share itself as the known-positive in the same run. All three fail on
cb0c10b with the GateUsage above.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1639 lines
66 KiB
Python
1639 lines
66 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 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), and all 15
|
|
# since the judge reads a passage as its exact reconstruction and one
|
|
# fixture delivers a passage (`set-passage.json`, one more unit), so M15
|
|
# has something to fell.
|
|
assert [row.number for row in rows if row.fails] == [5, 8]
|
|
assert (by_number[1].k, by_number[1].m) == (14, 14)
|
|
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) == (14, 14)
|
|
assert (by_number[7].k, by_number[7].m) == (15, 15)
|
|
|
|
|
|
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
|
|
|
|
|
|
# --- row 5 reads a hold-out set in the schema it declares --------------------
|
|
#
|
|
# The registered hold-out set is written in a consumer's own question-set
|
|
# schema, the one row 8 already reads, and row 5 used to read every set with
|
|
# the gate's synthetic reader: `not a question set this gate can read:
|
|
# 'set_id'`, a NO no set in that schema could ever turn. Every file below is
|
|
# written in that shape with INVENTED content over the synthetic bundle; the
|
|
# real hold-out set is never read by a test.
|
|
|
|
#: Distinctive on purpose: a row that printed a question's id or text would
|
|
#: carry one of these, and nothing else in the gate's output can.
|
|
_HOLD_OUT_MISS_ID = "HOLDOUT-MISS-7Q"
|
|
_HOLD_OUT_MISS_TEXT = "Hvilken farge har den fraflyttede fyrlykta paa Utvaer?"
|
|
_HOLD_OUT_HIT_IDS = ("HOLDOUT-H1", "HOLDOUT-H2", "HOLDOUT-H3", "HOLDOUT-H4")
|
|
|
|
|
|
def _fase_set(path: Path, *, schema: str = "fase-sporsmaal/1") -> Path:
|
|
"""Four questions the positive bundle answers and one it cannot: a
|
|
share of 4 of 5, counted here off the file and not off the row."""
|
|
hits = [
|
|
("Naar kontrolleres vinterberedskapen?", "haandbok", "innen 1. november"),
|
|
("Hvordan kvitteres noekkelen ut?", "haandbok", "signatur i noekkelboka"),
|
|
("Hvor ofte byttes batteriet i roekvarsleren?", "haandbok", "en gang i aaret"),
|
|
("Hva vedtok aarsmoetet om kontingenten?", "referat", "480 kroner"),
|
|
]
|
|
questions = [
|
|
{"id": qid, "question": text, "fasit": [{"doc": doc, "quote": quote}]}
|
|
for qid, (text, doc, quote) in zip(_HOLD_OUT_HIT_IDS, hits)
|
|
]
|
|
questions.append(
|
|
{
|
|
"id": _HOLD_OUT_MISS_ID,
|
|
"question": _HOLD_OUT_MISS_TEXT,
|
|
"fasit": [{"doc": "haandbok", "quote": "fyrlykta er malt i gult og fiolett"}],
|
|
}
|
|
)
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema": schema,
|
|
"hit_rule": "an excerpt whose source_file is <doc>.md AND carries the quote",
|
|
"questions": questions,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
def test_row_five_measures_a_hold_out_set_in_the_wiki_schema(tmp_path: Path) -> None:
|
|
"""Both directions through the same set and the same bundle: 4 of 5
|
|
clears 0.8 and falls under 0.9. The share is counted off the file first,
|
|
so the row is not the only thing that knows it."""
|
|
held_out = _fase_set(tmp_path / "held-out.json")
|
|
written = json.loads(held_out.read_text(encoding="utf-8"))
|
|
assert len(written["questions"]) == 5
|
|
bundle = _bundles(tmp_path)["positive"]
|
|
|
|
clears = gate.row_five(
|
|
_registration(tmp_path, set_path=held_out, bundle=bundle, threshold=0.8, name="c.json"),
|
|
provenance=lambda _: _carried_by_git(),
|
|
)
|
|
assert any(
|
|
"the measured hold-out clears the threshold: yes (4 of 5 = 0.8000 against 0.8000)" in detail
|
|
for detail in clears.details
|
|
), clears.details
|
|
assert (clears.k, clears.m, clears.status) == (11, 11, gate.GREEN)
|
|
|
|
falls = gate.row_five(
|
|
_registration(tmp_path, set_path=held_out, bundle=bundle, threshold=0.9, name="f.json"),
|
|
provenance=lambda _: _carried_by_git(),
|
|
)
|
|
assert any(
|
|
"the measured hold-out clears the threshold: NO (4 of 5 = 0.8000 against 0.9000)" in detail
|
|
for detail in falls.details
|
|
), falls.details
|
|
assert (falls.k, falls.m, falls.status) == (10, 11, gate.RED)
|
|
|
|
|
|
def test_a_hold_out_set_in_an_unknown_schema_is_a_named_no(tmp_path: Path) -> None:
|
|
"""Fail closed: a set read in a shape it was not written in measures
|
|
nothing, so an unknown schema is a NO that NAMES the schema -- never a
|
|
guess at a reader and never an exception."""
|
|
held_out = _fase_set(tmp_path / "held-out.json", schema="noe-annet/9")
|
|
row = gate.row_five(
|
|
_registration(
|
|
tmp_path, set_path=held_out, bundle=_bundles(tmp_path)["positive"], threshold=0.8
|
|
),
|
|
provenance=lambda _: _carried_by_git(),
|
|
)
|
|
verdict = [d for d in row.details if "the measured hold-out clears the threshold" in d]
|
|
assert len(verdict) == 1, row.details
|
|
assert ": NO (" in verdict[0]
|
|
assert "schema `noe-annet/9`" in verdict[0], verdict[0]
|
|
assert "fase-sporsmaal/1" in verdict[0], verdict[0]
|
|
assert row.fails
|
|
|
|
|
|
def test_row_five_prints_the_share_and_never_which_hold_out_question_missed(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A hold-out set whose misses are printed is a tuning set by the next
|
|
session. The known-positive in the same run: the row DID measure, one
|
|
question missed, and the share is printed -- so the absence of the ids
|
|
and the text below is the row's silence and not a row that never ran."""
|
|
held_out = _fase_set(tmp_path / "held-out.json")
|
|
for threshold in (0.8, 0.9):
|
|
row = gate.row_five(
|
|
_registration(
|
|
tmp_path,
|
|
set_path=held_out,
|
|
bundle=_bundles(tmp_path)["positive"],
|
|
threshold=threshold,
|
|
name=f"r{threshold}.json",
|
|
),
|
|
provenance=lambda _: _carried_by_git(),
|
|
)
|
|
text = gate.render([row]) + json.dumps(row.to_json(), ensure_ascii=False)
|
|
assert "4 of 5 = 0.8000" in text, text
|
|
assert _HOLD_OUT_MISS_ID not in text
|
|
assert _HOLD_OUT_MISS_TEXT not in text
|
|
assert "fyrlykta" not in text
|
|
for hit in _HOLD_OUT_HIT_IDS:
|
|
assert hit not in text
|