172 lines
6.7 KiB
Python
172 lines
6.7 KiB
Python
"""Bundle identity at Door B: the caller names the bundle, the door mirrors it.
|
|
|
|
Order `…2527032751` settled the form: **(c) a root-frontmatter bundle
|
|
identifier that consumers join on**. Two facts made this new wiring rather than
|
|
a value change. First, `root_frontmatter` had ZERO references in `inbox.py`:
|
|
Door B seeded its index with an empty file and no frontmatter block at all,
|
|
where Door A has carried the machinery since D5. Second, identity had to be
|
|
CALLER-owned — a bundle is a collection the caller delimits, so hashing its
|
|
contents would hand it a new identity every time a document was added.
|
|
|
|
That is also why there are two identity mechanisms in this library, stated
|
|
rather than hidden: WITHIN a bundle, Door C keys on a content hash; ACROSS
|
|
bundles, identity is this caller-assigned `bundle_id`. A consumer meets both.
|
|
|
|
S4b is resolved as ONE branch, not two: the root index is the SOURCE (the
|
|
caller supplies the value exactly once, D5 intact) and each concept MIRRORS
|
|
it. Two bundles built from identical inputs therefore hold concepts whose
|
|
paths COLLIDE by construction and whose identity values are disjoint — the
|
|
collision is the expected behaviour under form (c), not a defect.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from llm_ingestion_okf.errors import MaterializationError
|
|
from llm_ingestion_okf.inbox import GateDecision, process_inbox
|
|
from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_V1, STRUCTURED_V1
|
|
|
|
INGESTED_AT = "2026-07-25T12:00:00Z"
|
|
|
|
DOCUMENT = "# Brannkonsept\n\nSeksjonering etter N500.\n"
|
|
|
|
|
|
def gate(text: str) -> GateDecision:
|
|
return GateDecision(sanitized_text=text, disposition="warn")
|
|
|
|
|
|
def drop(inbox: Path, name: str, text: str = DOCUMENT) -> None:
|
|
inbox.mkdir(parents=True, exist_ok=True)
|
|
(inbox / name).write_text(text, encoding="utf-8", newline="")
|
|
|
|
|
|
def run(tmp: Path, *, profile=SEGMENTED_V1, values=None, round_name: str = "round"):
|
|
return process_inbox(
|
|
tmp / round_name,
|
|
tmp / "bundle",
|
|
INGESTED_AT,
|
|
okf_type="reference",
|
|
gate=gate,
|
|
profile=profile,
|
|
root_frontmatter_values=values,
|
|
)
|
|
|
|
|
|
def tree(bundle: Path) -> dict[str, bytes]:
|
|
"""Every file under the bundle, keyed by bundle-relative path."""
|
|
if not bundle.is_dir():
|
|
return {}
|
|
return {
|
|
str(path.relative_to(bundle)): path.read_bytes()
|
|
for path in sorted(bundle.rglob("*"))
|
|
if path.is_file()
|
|
}
|
|
|
|
|
|
# --- additivity: the four shipped profiles do not move --------------------
|
|
|
|
|
|
def test_default_without_values_writes_what_it_wrote_before(tmp_path: Path) -> None:
|
|
drop(tmp_path / "a", "n500-vegbygging.md")
|
|
process_inbox(
|
|
tmp_path / "a",
|
|
tmp_path / "old",
|
|
INGESTED_AT,
|
|
okf_type="reference",
|
|
gate=gate,
|
|
profile=DEFAULT,
|
|
)
|
|
drop(tmp_path / "b", "n500-vegbygging.md")
|
|
process_inbox(
|
|
tmp_path / "b",
|
|
tmp_path / "new",
|
|
INGESTED_AT,
|
|
okf_type="reference",
|
|
gate=gate,
|
|
profile=DEFAULT,
|
|
root_frontmatter_values=None,
|
|
)
|
|
assert tree(tmp_path / "new") == tree(tmp_path / "old")
|
|
assert not (tmp_path / "new" / "index.md").read_text(encoding="utf-8").startswith("---")
|
|
|
|
|
|
def test_structured_v1_without_values_still_writes_no_frontmatter(tmp_path: Path) -> None:
|
|
drop(tmp_path / "round", "n500-vegbygging.md")
|
|
run(tmp_path, profile=STRUCTURED_V1)
|
|
assert not (tmp_path / "bundle" / "index.md").read_text(encoding="utf-8").startswith("---")
|
|
|
|
|
|
# --- emission -------------------------------------------------------------
|
|
|
|
|
|
def test_the_root_index_opens_with_the_declared_bundle_id(tmp_path: Path) -> None:
|
|
drop(tmp_path / "round", "n500-vegbygging.md")
|
|
run(tmp_path, values={"bundle_id": "b-1"})
|
|
body = (tmp_path / "bundle" / "index.md").read_text(encoding="utf-8")
|
|
assert body.startswith("---\nbundle_id: b-1\n---\n\n")
|
|
|
|
|
|
def test_the_value_is_written_verbatim_and_unquoted(tmp_path: Path) -> None:
|
|
# Read as RAW BYTES: a parsed assertion returns the same string whether or
|
|
# not the emitter added quotes, so it would mask exactly the defect a
|
|
# consumer's shape gate trips on.
|
|
drop(tmp_path / "round", "n500-vegbygging.md")
|
|
run(tmp_path, values={"bundle_id": "svv-n500-2026"})
|
|
assert b"bundle_id: svv-n500-2026\n" in (tmp_path / "bundle" / "index.md").read_bytes()
|
|
|
|
|
|
def test_a_second_round_does_not_double_the_block(tmp_path: Path) -> None:
|
|
drop(tmp_path / "one", "n500-vegbygging.md")
|
|
run(tmp_path, values={"bundle_id": "b-1"}, round_name="one")
|
|
drop(tmp_path / "two", "v720-tunnel.md")
|
|
run(tmp_path, values={"bundle_id": "b-1"}, round_name="two")
|
|
body = (tmp_path / "bundle" / "index.md").read_text(encoding="utf-8")
|
|
assert body.startswith("---\nbundle_id: b-1\n---\n\n")
|
|
assert body.count("bundle_id:") == 1
|
|
|
|
|
|
def test_the_index_still_carries_its_entries_below_the_block(tmp_path: Path) -> None:
|
|
drop(tmp_path / "round", "n500-vegbygging.md")
|
|
run(tmp_path, values={"bundle_id": "b-1"})
|
|
body = (tmp_path / "bundle" / "index.md").read_text(encoding="utf-8")
|
|
assert "](inbox-n500-vegbygging.md)" in body
|
|
assert body.index("bundle_id") < body.index("](inbox-n500-vegbygging.md)")
|
|
|
|
|
|
# --- fail-fast, before any disk mutation ----------------------------------
|
|
|
|
|
|
def test_a_key_the_policy_does_not_name_is_refused_before_any_write(tmp_path: Path) -> None:
|
|
drop(tmp_path / "round", "n500-vegbygging.md")
|
|
before = tree(tmp_path / "bundle")
|
|
with pytest.raises(MaterializationError) as excinfo:
|
|
run(tmp_path, values={"okf_version": "0.2"})
|
|
assert excinfo.value.code == "index_root_frontmatter_unexpected"
|
|
# The whole point of hoisting the call above the write: a refused run must
|
|
# leave nothing behind, not a half-built bundle whose index is missing.
|
|
assert tree(tmp_path / "bundle") == before
|
|
assert before == {}
|
|
|
|
|
|
def test_a_refusal_leaves_an_existing_bundle_untouched(tmp_path: Path) -> None:
|
|
drop(tmp_path / "one", "n500-vegbygging.md")
|
|
run(tmp_path, values={"bundle_id": "b-1"}, round_name="one")
|
|
before = tree(tmp_path / "bundle")
|
|
assert before != {}
|
|
|
|
drop(tmp_path / "two", "v720-tunnel.md")
|
|
with pytest.raises(MaterializationError) as excinfo:
|
|
run(tmp_path, values={"okf_version": "0.2"}, round_name="two")
|
|
assert excinfo.value.code == "index_root_frontmatter_unexpected"
|
|
assert tree(tmp_path / "bundle") == before
|
|
|
|
|
|
def test_values_are_refused_against_a_profile_naming_no_root_keys(tmp_path: Path) -> None:
|
|
drop(tmp_path / "round", "n500-vegbygging.md")
|
|
with pytest.raises(MaterializationError) as excinfo:
|
|
run(tmp_path, profile=DEFAULT, values={"bundle_id": "b-1"})
|
|
assert excinfo.value.code == "index_root_frontmatter_unexpected"
|
|
assert tree(tmp_path / "bundle") == {}
|