"""The consumption pre-pass, checked rather than described. `tools/okf_consume.py` cuts an OKF bundle to one contract-conformant payload for one question. Three disciplines this suite is held to, all of them the house pattern rather than new inventions: - **Every zero carries a control.** A count of nothing is a measurement whose query must first be shown capable of finding. The placeholder scan runs against the template (known-positive) before its zero on the filled copy is believed; the index walk is controlled against the `rglob` the contract forbids the consumer path from using; the socket guard is fired directly before its silence during a real run counts as evidence. - **One mutation per rule.** `tests/test_contract_check.py` establishes the shape: assert the code a defect produces, never merely that something failed. - **The corpus is never a test dependency.** K2 lives outside the repository. Every test here runs against `examples/.../expected-bundle` (3 concepts) or `tests/fixtures/consume-bundle` (synthetic, carrying the states the real corpus has zero of). The corpus-conditional arm skips with its denominator named, so a skip cannot read as a pass. """ from __future__ import annotations import hashlib import json import os import sys import unicodedata from pathlib import Path import pytest PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT / "tools")) import okf_consume # noqa: E402 from llm_ingestion_okf.materialize import parse_frontmatter # noqa: E402 GOLDEN = PROJECT_ROOT / "examples" / "ingest-golden-segmented-okf-v0-2" / "expected-bundle" # --- Step 1: the walk and the ref --------------------------------------------- def test_the_index_walk_finds_every_concept_and_no_index_or_log() -> None: found = okf_consume.enumerate_concepts(GOLDEN) assert found == ( "krav/1-1/foerste-krav", "krav/1-2/andre-krav", "veiledning", ) assert not any(concept.endswith("index") or concept.endswith("log") for concept in found) def test_the_index_walk_is_complete_against_the_method_the_contract_forbids() -> None: # SS 9.2 forbids the CONSUMER path from enumerating a directory. Using the # forbidden method here, in a test, is what proves the permitted one loses # nothing -- an absence with no control is not a measurement. by_rglob = { path.relative_to(GOLDEN).with_suffix("").as_posix() for path in GOLDEN.rglob("*.md") if path.name not in ("index.md", "log.md") } assert by_rglob, "the control found nothing, so it cannot certify the walk" assert set(okf_consume.enumerate_concepts(GOLDEN)) == by_rglob def test_a_concept_reachable_only_through_a_nested_index_is_still_found() -> None: # `krav/1-1/foerste-krav` is three levels down and named in no root entry. root_entries = (GOLDEN / "index.md").read_text(encoding="utf-8") assert "foerste-krav" not in root_entries, "the fixture no longer exercises nesting" assert "krav/1-1/foerste-krav" in okf_consume.enumerate_concepts(GOLDEN) def test_the_ref_is_stable_across_calls_and_names_its_algorithm() -> None: first = okf_consume.bundle_ref(GOLDEN) assert first == okf_consume.bundle_ref(GOLDEN) assert first.startswith("sha256-tree:") def test_the_ref_moves_when_one_concept_byte_moves(tmp_path: Path) -> None: copy = tmp_path / "bundle" _copy_bundle(GOLDEN, copy) before = okf_consume.bundle_ref(copy) target = copy / "veiledning.md" target.write_text(target.read_text(encoding="utf-8") + "x", encoding="utf-8") assert okf_consume.bundle_ref(copy) != before def test_the_ref_does_not_move_when_only_mtimes_move(tmp_path: Path) -> None: copy = tmp_path / "bundle" _copy_bundle(GOLDEN, copy) before = okf_consume.bundle_ref(copy) for path in sorted(copy.rglob("*")): if path.is_file(): os.utime(path, (0, 0)) assert okf_consume.bundle_ref(copy) == before def _copy_bundle(source: Path, target: Path) -> None: for path in sorted(source.rglob("*")): if path.is_file(): destination = target / path.relative_to(source) destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(path.read_bytes()) def test_the_ref_covers_the_indexes_too_since_the_walk_reads_them(tmp_path: Path) -> None: # The docstring claims every byte that can reach a payload is inside the # ref. An index byte can: it decides which concepts are reachable at all. copy = tmp_path / "bundle" _copy_bundle(GOLDEN, copy) before = okf_consume.bundle_ref(copy) nested = copy / "krav" / "1-1" / "index.md" nested.write_text(nested.read_text(encoding="utf-8") + "\nfritekst\n", encoding="utf-8") assert okf_consume.bundle_ref(copy) != before # --- Step 2: one concept, read into a record ---------------------------------- PROPOSED_CONCEPT = "krav/1-1/foerste-krav" ROOT_BUNDLE_ID = "b-golden-segmented-okf-v0-2" def _read(concept_id: str, root: Path = GOLDEN) -> okf_consume.Concept: return okf_consume.read_concept( root / f"{concept_id}.md", bundle_root=root, root_bundle_id=ROOT_BUNDLE_ID ) def test_a_concept_carrying_adjudication_reads_that_value() -> None: assert _read(PROPOSED_CONCEPT).adjudication == "proposed" def test_a_concept_carrying_no_adjudication_key_reads_as_unknown(tmp_path: Path) -> None: # SS 6.1: `unknown` is written EXPLICITLY. "Not judged" and "we cannot tell # whether it was judged" are different facts, and only one is about the # concept. root = tmp_path / "bundle" _copy_bundle(GOLDEN, root) target = root / f"{PROPOSED_CONCEPT}.md" target.write_text( target.read_text(encoding="utf-8").replace("adjudication: proposed\n", ""), encoding="utf-8", ) concept = _read(PROPOSED_CONCEPT, root) assert concept.adjudication == "unknown" assert concept.adjudication_present is False def test_an_adjudication_value_outside_the_wire_set_is_refused_by_name(tmp_path: Path) -> None: # Mapping an unrecognised value to `unknown` would report "we cannot tell" # where the truth is "the bundle said something this consumer does not # understand" -- a defect laundered into a state. root = tmp_path / "bundle" _copy_bundle(GOLDEN, root) target = root / f"{PROPOSED_CONCEPT}.md" target.write_text( target.read_text(encoding="utf-8").replace("adjudication: proposed", "adjudication: seen"), encoding="utf-8", ) with pytest.raises(okf_consume.ConsumeError) as raised: _read(PROPOSED_CONCEPT, root) assert raised.value.code == "adjudication_unknown_value" def test_the_digest_is_of_the_concept_file_and_is_not_the_source_sha256() -> None: concept = _read(PROPOSED_CONCEPT) on_disk = hashlib.sha256((GOLDEN / f"{PROPOSED_CONCEPT}.md").read_bytes()).hexdigest() assert concept.sha256 == on_disk assert len(concept.sha256) == 64 frontmatter = parse_frontmatter(GOLDEN / f"{PROPOSED_CONCEPT}.md") assert frontmatter["source_sha256"] != concept.sha256 def test_the_concept_id_keeps_its_slashes_where_import_slug_would_flatten_them() -> None: # `importer.import_slug` flattens one line below the rule this id follows. # A flattened id fails a document-prefix match in a way that looks like a # ranking miss rather than an id-format bug. assert _read(PROPOSED_CONCEPT).concept_id == "krav/1-1/foerste-krav" def test_bundle_id_falls_back_to_the_root_index_and_says_that_it_did(tmp_path: Path) -> None: root = tmp_path / "bundle" _copy_bundle(GOLDEN, root) target = root / f"{PROPOSED_CONCEPT}.md" target.write_text( target.read_text(encoding="utf-8").replace(f"bundle_id: {ROOT_BUNDLE_ID}\n", ""), encoding="utf-8", ) concept = _read(PROPOSED_CONCEPT, root) assert concept.bundle_id == ROOT_BUNDLE_ID assert concept.bundle_id_inherited is True assert _read(PROPOSED_CONCEPT).bundle_id_inherited is False # --- Step 3: trust_tier, and the refusal to tier what cannot be read ---------- FIXTURE = PROJECT_ROOT / "tests" / "fixtures" / "consume-bundle" def test_the_fixture_bundle_carries_what_the_real_corpus_has_none_of() -> None: # The control on every assertion below. K2 has 0 `verified:` keys, 0 # `type: verdict` and 0 `adjudication: adjudicated` over 629 concepts, so a # fixture missing any of them would make its tests pass over an empty set. text = "\n".join(path.read_text(encoding="utf-8") for path in sorted(FIXTURE.rglob("*.md"))) assert "type: verdict" in text assert "adjudication: adjudicated" in text assert "by: human:" in text assert "by: process:" in text def test_no_verified_key_reads_as_unverified() -> None: # SS 6.3: a concept carrying no trust frontmatter is still consumable. assert okf_consume.trust_tier(None) == "unverified" def test_a_human_actor_reads_as_human_reviewed() -> None: assert okf_consume.trust_tier("[{ by: human:ktg, at: 2026-09-01T00:00:00Z }]") == ( "human-reviewed" ) def test_a_process_actor_reads_as_machine_confirmed() -> None: assert okf_consume.trust_tier("[{ by: process:okf-check, at: 2026-09-01T00:00:00Z }]") == ( "machine-confirmed" ) def test_the_human_test_is_a_prefix_and_never_a_substring() -> None: # `bot/human:2` is a MACHINE actor whose id contains the string `human:`. # A substring test would promote it to the highest tier -- fabricated # provenance produced by a matching bug. assert okf_consume.trust_tier("[{ by: bot/human:2 }]") == "machine-confirmed" def test_an_entry_naming_no_actor_is_refused_rather_than_tiered() -> None: with pytest.raises(okf_consume.ConsumeError) as raised: okf_consume.trust_tier("[{ at: 2026-09-01T00:00:00Z }]") assert raised.value.code == "verified_actorless" def test_a_block_form_verified_is_not_a_tier_at_all() -> None: # Measured 2026-09-07: this library's line-oriented `parse_frontmatter` # returns `''` for a block-form `verified:` and the full string for a flow # one, so PRESENT-BUT-UNREADABLE is distinguishable from ABSENT. Emitting # `unverified` here would assert a fact nobody measured (SS 6.4). assert okf_consume.trust_tier("") is None def test_the_block_form_case_is_real_in_the_fixture_and_not_only_in_the_unit_test() -> None: # The control: without this, `trust_tier("") is None` could be true of a # string no bundle ever produces. frontmatter = parse_frontmatter(FIXTURE / "dyp" / "nivaa" / "blokkform-verifisert.md") assert frontmatter["verified"] == "" assert okf_consume.trust_tier(frontmatter["verified"]) is None # --- Step 4: the budget instrument ------------------------------------------- CONTRACT = PROJECT_ROOT / "docs" / "consumption-contract.md" def test_measure_counts_bytes_and_not_characters() -> None: # The exact conflation the brief records itself making once: a chars/token # ratio quoted where a bytes/token one was needed. `æøå` is three # characters and six bytes, and the two only differ outside ASCII. assert okf_consume.measure("æøå") == len('"æøå"'.encode()) assert okf_consume.measure("æøå") != len("æøå") def test_measure_counts_the_encoded_form_the_payload_actually_costs() -> None: # `json.dumps` defaults to `ensure_ascii=True`, which inflates this corpus # by 7.1 %. A gate measuring one form while the knapsack weighs the other # disagrees by more than the headroom. norwegian = "årlig kontroll av anlegget" assert okf_consume.measure(norwegian) == len( json.dumps(norwegian, ensure_ascii=False).encode("utf-8") ) assert okf_consume.measure(norwegian) < len( json.dumps(norwegian, ensure_ascii=True).encode("utf-8") ) def test_the_known_positive_is_reproduced_by_the_gates_own_instrument() -> None: case, expected, measured = okf_consume.known_positive() assert case assert expected == measured, "SS 7.4: the instrument has not been shown to count" def test_the_known_positive_is_not_the_raw_byte_count_of_the_same_file() -> None: # Validating one instrument while gating with another is the SS 7.4 failure # the rule exists to prevent. The delta is derivable by a second, wholly # independent route (`wc -c`) and moves the moment `measure` changes what it # counts -- which is what keeps `expected == measured` from being vacuous. _, expected, _ = okf_consume.known_positive() raw = len(CONTRACT.read_bytes()) assert expected != raw assert expected - raw == okf_consume.KNOWN_POSITIVE_ENCODING_DELTA def test_the_default_limit_admits_a_concept_the_size_of_the_price_form() -> None: # Measured during planning: at the drafted 60 000 B default the SC6 gold # concept (101 313 B encoded) falls to the "cannot fit alone" pre-exclusion, # so SC1 and SC6 were mutually unsatisfiable on a CORRECT implementation. assert okf_consume.DEFAULT_LIMIT >= 101_313 def test_the_budget_unit_and_instrument_are_named_rather_than_implied() -> None: assert "byte" in okf_consume.BUDGET_UNIT assert "ensure_ascii=False" in okf_consume.BUDGET_INSTRUMENT # --- Step 5: stage-one document ranking -------------------------------------- def test_normalise_is_nfc_stable_on_the_one_letter_that_decomposes() -> None: # `NFD("å")` is `a` + U+030A, and the combining ring is not `\w`, so an # un-normalised split returns `["a", "rlig"]`. `æ` and `ø` have NO canonical # decomposition, so a test built on `miljø` passes while the bug is live -- # the known-positive here MUST use `å`. composed = unicodedata.normalize("NFC", "årlig kontroll") decomposed = unicodedata.normalize("NFD", "årlig kontroll") assert composed != decomposed, "the control is broken: the two forms are identical" assert okf_consume.normalise(decomposed) == okf_consume.normalise(composed) assert "årlig" in okf_consume.normalise(decomposed) def test_normalise_drops_tokens_under_three_characters() -> None: assert okf_consume.normalise("er en pris i et skjema") == ("pris", "skjema") def test_two_tokens_match_on_a_shared_prefix_of_four_and_not_of_three() -> None: # "Stem-substring" is not an implementable rule: neither `prisene` nor # `prissammenstilling` contains the other. Shared prefix does the work -- # `pris|ene` and `pris|sammenstilling` share 4. A 3-character floor # over-matches Norwegian function words. assert okf_consume.tokens_match("prisene", "prissammenstilling") assert okf_consume.tokens_match("kontrollen", "kontroll") assert not okf_consume.tokens_match("pris", "pri") assert not okf_consume.tokens_match("krav", "kraft") def test_a_question_naming_a_directorys_subject_ranks_that_directory_first() -> None: scores = okf_consume.document_scores(FIXTURE, "Hvordan skal prisene fylles ut?") assert scores, "no document scored, so 'ranks first' would measure nothing" assert max(scores, key=lambda key: (scores[key], key)) == "krav" def test_a_question_about_a_different_subject_ranks_a_different_directory() -> None: # The control on the test above: without it, a scorer returning "krav" # unconditionally would pass. scores = okf_consume.document_scores(FIXTURE, "Hva er omfanget og formaalet?") assert max(scores, key=lambda key: (scores[key], key)) == "scope" def test_curated_prose_in_an_index_is_ignored_rather_than_scored(tmp_path: Path) -> None: root = tmp_path / "bundle" _copy_bundle(FIXTURE, root) index = root / "krav" / "index.md" index.write_text( "Denne mappen handler om priser og prissammenstilling.\n\n" + index.read_text(encoding="utf-8"), encoding="utf-8", ) assert ( okf_consume.DEFAULT_PROFILE.index.parse_entry( "Denne mappen handler om priser og prissammenstilling." ) is None ) assert okf_consume.document_scores(root, "Hvordan skal prisene fylles ut?") == ( okf_consume.document_scores(FIXTURE, "Hvordan skal prisene fylles ut?") ) def test_document_scores_are_identical_across_two_calls() -> None: question = "Hvordan skal prisene fylles ut?" assert okf_consume.document_scores(FIXTURE, question) == okf_consume.document_scores( FIXTURE, question ) # --- Step 6: stage-two concept ranking, fused by RRF -------------------------- def _fixture_concepts() -> list[okf_consume.Concept]: return [ okf_consume.read_concept( FIXTURE / f"{concept_id}.md", bundle_root=FIXTURE, root_bundle_id="consume-fixture", ) for concept_id in okf_consume.enumerate_concepts(FIXTURE) ] def test_concepts_tying_on_every_signal_come_back_in_concept_id_order() -> None: concepts = _fixture_concepts() # A question matching nothing makes every signal identical, so the ONLY # thing left deciding the order is the declared tie-break. ranked = okf_consume.concept_scores(concepts, "zzzz qqqq", {}) ids = [concept.concept_id for concept, _ in ranked] assert ids == sorted(ids) def test_reversing_the_input_order_does_not_change_the_output_order() -> None: concepts = _fixture_concepts() forward = [c.concept_id for c, _ in okf_consume.concept_scores(concepts, "zzzz qqqq", {})] backward = [ c.concept_id for c, _ in okf_consume.concept_scores(list(reversed(concepts)), "zzzz qqqq", {}) ] assert forward == backward def test_a_concept_in_a_high_scoring_document_outranks_an_equally_lexical_one() -> None: concepts = _fixture_concepts() question = "Hvordan skal prisene fylles ut?" lifted = okf_consume.concept_scores(concepts, question, {"krav": 10.0, "dyp": 0.0}) dropped = okf_consume.concept_scores(concepts, question, {"krav": 0.0, "dyp": 10.0}) krav_first = [c.concept_id for c, _ in lifted].index("krav/prissammenstilling") krav_later = [c.concept_id for c, _ in dropped].index("krav/prissammenstilling") assert krav_first < krav_later def test_the_ranked_order_is_identical_across_two_calls() -> None: concepts = _fixture_concepts() scores = okf_consume.document_scores(FIXTURE, "Hvordan skal prisene fylles ut?") first = okf_consume.concept_scores(concepts, "Hvordan skal prisene fylles ut?", scores) second = okf_consume.concept_scores(concepts, "Hvordan skal prisene fylles ut?", scores) assert [c.concept_id for c, _ in first] == [c.concept_id for c, _ in second] def test_the_price_concept_leads_on_the_price_question_in_the_fixture() -> None: concepts = _fixture_concepts() scores = okf_consume.document_scores(FIXTURE, "Hvordan skal prisene fylles ut?") ranked = okf_consume.concept_scores(concepts, "Hvordan skal prisene fylles ut?", scores) assert ranked[0][0].concept_id == "krav/prissammenstilling"