110 lines
4.6 KiB
Python
110 lines
4.6 KiB
Python
"""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 os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(PROJECT_ROOT / "tools"))
|
|
|
|
import okf_consume # 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
|