llm-ingestion-okf/tests/test_okf_consume.py

"""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 os
import sys
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