"""The measuring instrument of `tools/okf_soek_gate.py`, over a synthetic corpus. The gate itself is NOT in this suite: it is red by construction against a real collection, and a red gate in a green suite is a suite nobody trusts. What is here is the part that can lie quietly -- the hit rule, the counting, the missing-fixture state and the merge -- measured against a corpus this file builds, where every answer is forced by the fixture rather than by a ranking. The synthetic sets are written in the SAME shapes the real ones carry, so the code path is measured even on a machine where no real set exists. """ from __future__ import annotations import json import sys from pathlib import Path import pytest TOOLS = Path(__file__).resolve().parent.parent / "tools" if str(TOOLS) not in sys.path: sys.path.insert(0, str(TOOLS)) import okf_retrieval_gate as retrieval # noqa: E402 import okf_soek_gate as gate # noqa: E402 from llm_ingestion_okf import consume # noqa: E402 # --- the synthetic corpus ----------------------------------------------------- # # Reuses the retrieval gate's bundle writer rather than a second one: two # writers would let the two gates' fixtures drift apart in shape, and the shape # is the thing both of them depend on. QUOTE_DELIVERED = "the roof beam is replaced every twelve years" QUOTE_PRESENT_NOT_DELIVERED = "the cellar key is kept by the treasurer" QUOTE_ABSENT = "the gondola runs on alternating tuesdays" SPEC = retrieval.BundleSpec( "soek-synthetic", ( retrieval.DocumentSpec( "handbook", "handbook.md", ( retrieval.ConceptSpec( slug="roof", title="Roof maintenance", body=f"The club inspects the cabin roof. {QUOTE_DELIVERED}.", ), ), ), retrieval.DocumentSpec( "cellar", "cellar.md", ( retrieval.ConceptSpec( slug="keys", title="Key holding", # Carries none of the roof question's words, so it is in the # collection and unreachable from that question: class # `soekefeil` is forced, not hoped for. body=f"Storage arrangements. {QUOTE_PRESENT_NOT_DELIVERED}.", ), ), ), ), ) @pytest.fixture(scope="module") def bundle(tmp_path_factory: pytest.TempPathFactory) -> Path: root = tmp_path_factory.mktemp("soek") / "bundle" return retrieval.build_bundle(root, SPEC) def write_sets(directory: Path, **files: object) -> Path: directory.mkdir(parents=True, exist_ok=True) for name, payload in files.items(): (directory / f"{name}.json").write_text( json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) return directory FASE = { "schema": "fase-sporsmaal/1", "frozen": "synthetic", "hit_rule": "source_file == .md AND the excerpt contains the quote.", "questions": [ { "id": "S1", "class": "docs", "question": "roof beam replacement interval", "fasit": [{"doc": "handbook", "quote": QUOTE_DELIVERED}], }, { "id": "S2", "class": "release_only", "question": "roof beam replacement interval", # In the collection, unreachable from this question's words. "fasit": [{"doc": "cellar", "quote": QUOTE_PRESENT_NOT_DELIVERED}], }, { "id": "S3", "class": "release_only", "question": "roof beam replacement interval", "fasit": [{"doc": "handbook", "quote": QUOTE_ABSENT}], }, ], } # --- the hit rule ------------------------------------------------------------- def test_the_hit_rule_needs_the_source_AND_the_quote() -> None: """Either half alone is not a hit -- the sets' own rule, verbatim.""" right = {"source_file": "handbook.md", "text": f"x {QUOTE_DELIVERED} y"} wrong_source = {"source_file": "cellar.md", "text": f"x {QUOTE_DELIVERED} y"} wrong_quote = {"source_file": "handbook.md", "text": "x nothing of the sort y"} assert gate.excerpt_carries(right, "handbook", QUOTE_DELIVERED) assert not gate.excerpt_carries(wrong_source, "handbook", QUOTE_DELIVERED) assert not gate.excerpt_carries(wrong_quote, "handbook", QUOTE_DELIVERED) def test_the_hit_rule_folds_case_and_collapses_whitespace() -> None: excerpt = {"source_file": "handbook.md", "text": "The ROOF\n beam\tis replaced"} assert gate.excerpt_carries(excerpt, "handbook", "the roof beam is replaced") def test_any_fasit_entry_suffices() -> None: """Three entries, one match: a hit. The rule says so in both real sets.""" excerpts = [{"source_file": "handbook.md", "text": QUOTE_DELIVERED}] fasit = [ {"doc": "cellar", "quote": QUOTE_ABSENT}, {"doc": "handbook", "quote": QUOTE_DELIVERED}, {"doc": "handbook", "quote": QUOTE_ABSENT}, ] assert gate.question_hit(excerpts, fasit) assert not gate.question_hit(excerpts, fasit[:1]) # --- counting and miss classification ---------------------------------------- def test_counting_and_miss_classes_over_the_synthetic_corpus(bundle: Path, tmp_path: Path) -> None: """One hit, one search failure, one build failure -- each forced by the fixture.""" sets = gate.load_sets(write_sets(tmp_path / "sets", **{"fase-sporsmaal": FASE})) report = gate.run(bundle, sets) row = report.row("a") assert (row.measured, row.denominator) == (1, 3) classes = {miss.question_id: miss.klass for miss in row.misses} assert classes == {"S2": gate.SEARCH_FAILURE, "S3": gate.BUILD_FAILURE} def test_a_miss_is_a_build_failure_only_when_no_fasit_quote_is_in_the_collection( bundle: Path, ) -> None: """The denominator of 'is it even there' is the collection, not the payload.""" text = gate.collection_text(bundle) assert gate.collapse(QUOTE_PRESENT_NOT_DELIVERED) in text["cellar.md"] assert gate.collapse(QUOTE_ABSENT) not in text.get("handbook.md", "") def test_release_only_counts_within_the_phase_set(bundle: Path, tmp_path: Path) -> None: """Series (b) is a subset of (a), never a second set and never a second ask.""" sets = gate.load_sets(write_sets(tmp_path / "sets", **{"fase-sporsmaal": FASE})) report = gate.run(bundle, sets) assert report.row("b").denominator == 2 assert report.row("b").measured == 0 # --- the missing fixture ------------------------------------------------------ def test_a_missing_fixture_is_red_and_never_zero_hits(bundle: Path, tmp_path: Path) -> None: """The distinction the order names: not run is not the same fact as no hits.""" sets = gate.load_sets(write_sets(tmp_path / "sets", **{"fase-sporsmaal": FASE})) report = gate.run(bundle, sets) for key in ("c", "d", "e", "f", "g1"): row = report.row(key) assert row.measured is None, key assert row.holds() is None, key assert gate.MISSING_FIXTURE in row.render(), key assert report.exit_code() == 1 def test_an_unreadable_fixture_is_wrong_input_not_a_red_row(tmp_path: Path) -> None: directory = tmp_path / "sets" directory.mkdir() (directory / "fase-sporsmaal.json").write_text("{not json", encoding="utf-8") with pytest.raises(gate.GateUsage): gate.load_sets(directory) def test_a_missing_collection_exits_two(tmp_path: Path) -> None: code = gate.main(["--bundle", str(tmp_path / "nowhere"), "--sets", str(tmp_path)]) assert code == 2 # --- the merge ---------------------------------------------------------------- def test_the_gate_has_no_merge_of_its_own() -> None: """Since v1.1 C2 the sub-questions of series (e) and (f) go to the product in ONE call, so the merge measured is the one a reader receives.""" assert not hasattr(gate, "merge_round_robin") def test_the_gate_asks_every_subquestion_in_one_call(bundle: Path) -> None: parts = ["roof beam replacement interval", "storage arrangements"] assert gate.Asker(bundle).many(parts) == consume.build_multi_payload(bundle, questions=parts) # --- the thresholds and the verdict ------------------------------------------ def test_every_threshold_is_a_named_constant() -> None: """A bar read off a literal in a row is a bar nobody can find again.""" for name in ( "THRESHOLD_PHASE", "THRESHOLD_RELEASE_ONLY", "THRESHOLD_HOLDOUT", "THRESHOLD_NORWEGIAN_DIRECT", "THRESHOLD_NORWEGIAN_SUBQUESTIONS", "THRESHOLD_OPERATOR", "THRESHOLD_NEGATIVE_FLAGGED", "THRESHOLD_POSITIVE_MISFLAGGED", "THRESHOLD_LARGEST_EXCERPT", ): assert isinstance(getattr(gate, name), int) def test_a_row_at_the_bar_holds_and_one_under_it_does_not() -> None: at = gate.Row("x", "x", measured=3, denominator=5, threshold=3) under = gate.Row("x", "x", measured=2, denominator=5, threshold=3) assert at.holds() and not under.holds() def test_an_at_most_row_reads_the_other_way() -> None: at = gate.Row("x", "x", measured=2, denominator=7, threshold=2, at_most=True) over = gate.Row("x", "x", measured=3, denominator=7, threshold=2, at_most=True) assert at.holds() and not over.holds() def test_exit_zero_needs_every_row(bundle: Path, tmp_path: Path) -> None: """One row short of its bar is exit 1, whichever row it is.""" sets = gate.load_sets(write_sets(tmp_path / "sets", **{"fase-sporsmaal": FASE})) report = gate.run(bundle, sets) assert report.exit_code() == 1 green = gate.Report( collection="x", rows=[gate.Row("a", "a", measured=1, denominator=1, threshold=1)], notes=(), ) assert green.exit_code() == 0 # --- determinism -------------------------------------------------------------- def test_two_renderings_are_byte_identical(bundle: Path, tmp_path: Path) -> None: sets = gate.load_sets(write_sets(tmp_path / "sets", **{"fase-sporsmaal": FASE})) first = gate.run(bundle, sets).render() second = gate.run(bundle, sets).render() assert first.encode("utf-8") == second.encode("utf-8") def test_the_rendering_carries_no_absolute_path(bundle: Path, tmp_path: Path) -> None: """The table is pasted into STATE and a commit; a scratch path in it is noise.""" sets = gate.load_sets(write_sets(tmp_path / "sets", **{"fase-sporsmaal": FASE})) rendered = gate.run(bundle, sets).render() assert str(bundle) not in rendered assert not any(line.strip().startswith("/") for line in rendered.splitlines()) # --- the negative signal ------------------------------------------------------ def test_the_negative_signal_is_the_retrieval_gates_reading() -> None: """One repository, one definition of 'the payload says it does not know'.""" assert gate.uncovered_signal is retrieval.marked