349 lines
13 KiB
Python
349 lines
13 KiB
Python
"""One document becomes many concepts: Door B's 1-to-N path.
|
|
|
|
The measured defect this closes. `process_inbox` emitted exactly one flat
|
|
concept per dropped file, which is the shape OKF v0.2's Appendix A presents
|
|
v0.1 as migrating AWAY from -- and §11 could never catch it, because it checks
|
|
that every non-reserved `.md` parses with a non-empty `type`. A bundle of one
|
|
giant concept is fully conformant. Conformance is the floor, not the proof.
|
|
|
|
Two invariants are load-bearing here and are asserted rather than assumed:
|
|
|
|
- **Every segment is gated BEFORE any segment is written.** Gating and writing
|
|
one at a time would leave a half-screened document on disk the moment
|
|
segment 3 of 5 quarantines -- part of a document the guard refused, persisted
|
|
and indexed, with the run reporting success.
|
|
- **A span indexes the CANONICAL EXTRACTED TEXT.** The `.csv` fixture below is
|
|
what proves it: `render_table` re-renders those bytes, so a span computed
|
|
against the source bytes lands on different characters. On a `.md` fixture
|
|
bytes and text are identical and the assertion would pass for the wrong
|
|
reason.
|
|
|
|
`csv` is a CORE type, needing no `[extract]` extra, so that fixture can never
|
|
skip. A `pdf` would be the trap: pdfplumber is installed in this venv, so it
|
|
would pass here and skip silently in a bare consumer environment.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from llm_ingestion_okf.errors import SegmentationError
|
|
from llm_ingestion_okf.extract import extract_text
|
|
from llm_ingestion_okf.inbox import GateDecision, process_inbox
|
|
from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_V1
|
|
from llm_ingestion_okf.segmentation import SegmentationPlan, parse_segmentation_plan
|
|
|
|
INGESTED_AT = "2026-07-25T12:00:00Z"
|
|
PLAN_AT = "2026-08-30T09:00:00Z"
|
|
|
|
DOCUMENT = (
|
|
"0 Forord: bakgrunn for konseptet.\n"
|
|
"1 Brannkonsept: krav til seksjonering.\n"
|
|
"2 Roemning: to uavhengige veier.\n"
|
|
"3 Baereevne: R60 for hovedbaeresystem.\n"
|
|
"4 Slokkeanlegg: sprinkler i hele bygget.\n"
|
|
)
|
|
|
|
PATHS = (
|
|
"krav/forord.md",
|
|
"krav/3-1/brannkonsept.md",
|
|
"krav/3-1/roemning.md",
|
|
"krav/3-2/baereevne.md",
|
|
"krav/3-2/slokkeanlegg.md",
|
|
)
|
|
|
|
|
|
def gate(text: str) -> GateDecision:
|
|
return GateDecision(sanitized_text=text, disposition="warn")
|
|
|
|
|
|
def drop(inbox: Path, name: str, text: str) -> Path:
|
|
inbox.mkdir(parents=True, exist_ok=True)
|
|
path = inbox / name
|
|
path.write_text(text, encoding="utf-8", newline="")
|
|
return path
|
|
|
|
|
|
def line_spans(text: str) -> tuple[tuple[int, int], ...]:
|
|
"""Half-open spans, one per line, measured rather than hard-coded."""
|
|
spans: list[tuple[int, int]] = []
|
|
offset = 0
|
|
for line in text.splitlines(keepends=True):
|
|
spans.append((offset, offset + len(line)))
|
|
offset += len(line)
|
|
return tuple(spans)
|
|
|
|
|
|
def build_plan(
|
|
source_bytes: bytes,
|
|
text: str,
|
|
*,
|
|
paths: tuple[str, ...] = PATHS,
|
|
extractor_id: str = "md",
|
|
**overrides: Any,
|
|
) -> SegmentationPlan:
|
|
spans = line_spans(text)
|
|
payload: dict[str, Any] = {
|
|
"version": "1",
|
|
"source_sha256": hashlib.sha256(source_bytes).hexdigest(),
|
|
"extractor_id": extractor_id,
|
|
"extractor_version": "1.0.0",
|
|
"adjudicated_at": "2026-08-30T08:00:00Z",
|
|
"entries": [
|
|
{
|
|
"segment_id": f"s{index}",
|
|
"path": path,
|
|
"title": f"Del {index}",
|
|
"okf_type": "requirement",
|
|
"span": list(spans[index]),
|
|
"ingested_at": PLAN_AT,
|
|
}
|
|
for index, path in enumerate(paths)
|
|
],
|
|
}
|
|
payload.update(overrides)
|
|
return parse_segmentation_plan(payload)
|
|
|
|
|
|
def run(
|
|
tmp: Path,
|
|
*,
|
|
plan: SegmentationPlan | None,
|
|
profile=SEGMENTED_V1,
|
|
values: dict[str, str] | None = None,
|
|
round_name: str = "round",
|
|
guard=gate,
|
|
):
|
|
return process_inbox(
|
|
tmp / round_name,
|
|
tmp / "bundle",
|
|
INGESTED_AT,
|
|
okf_type="requirement",
|
|
gate=guard,
|
|
profile=profile,
|
|
root_frontmatter_values={"bundle_id": "b-1"} if values is None else values,
|
|
segmentation=plan,
|
|
)
|
|
|
|
|
|
def concepts(bundle: Path, profile=SEGMENTED_V1) -> dict[str, str]:
|
|
"""Every non-reserved concept, keyed by bundle-relative path.
|
|
|
|
Reserved names are excluded at EVERY level, not just the root -- a
|
|
per-directory index writer puts one in each directory, and counting those
|
|
as concepts would inflate the count the moment nesting appeared.
|
|
"""
|
|
if not bundle.is_dir():
|
|
return {}
|
|
return {
|
|
str(path.relative_to(bundle)): path.read_text(encoding="utf-8")
|
|
for path in sorted(bundle.rglob(f"*{profile.paths.concept_suffix}"))
|
|
if path.is_file() and path.name != profile.index.name
|
|
}
|
|
|
|
|
|
def tree(bundle: Path) -> dict[str, bytes]:
|
|
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()
|
|
}
|
|
|
|
|
|
def body_of(document: str) -> str:
|
|
return document.split("---\n", 2)[2].lstrip("\n")
|
|
|
|
|
|
def frontmatter_of(document: str) -> dict[str, str]:
|
|
head = document.split("---\n")[1]
|
|
return dict(
|
|
line.split(": ", 1) for line in head.splitlines() if ": " in line and line[:1] != " "
|
|
)
|
|
|
|
|
|
# --- S1: one document, exactly N concepts ---------------------------------
|
|
|
|
|
|
def test_a_five_entry_plan_yields_exactly_five_concepts(tmp_path: Path) -> None:
|
|
source = drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
run(tmp_path, plan=build_plan(source.read_bytes(), DOCUMENT))
|
|
found = concepts(tmp_path / "bundle")
|
|
assert len(found) == 5
|
|
assert len(found) > 1
|
|
assert set(found) == set(PATHS)
|
|
|
|
|
|
def test_without_a_plan_the_same_fixture_yields_exactly_one_concept(tmp_path: Path) -> None:
|
|
# The discriminating negative control. Without it, a test that counts five
|
|
# concepts proves nothing about whether the PLAN caused the split.
|
|
drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
run(tmp_path, plan=None)
|
|
found = concepts(tmp_path / "bundle")
|
|
assert len(found) == 1
|
|
assert set(found) == {"inbox-n500.md"}
|
|
|
|
|
|
def test_each_segment_id_maps_to_the_path_its_entry_declares(tmp_path: Path) -> None:
|
|
source = drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
plan = build_plan(source.read_bytes(), DOCUMENT)
|
|
run(tmp_path, plan=plan)
|
|
found = concepts(tmp_path / "bundle")
|
|
mapped = {frontmatter_of(document)["segment_id"]: path for path, document in found.items()}
|
|
assert mapped == {item.segment_id: item.path for item in plan.entries}
|
|
|
|
|
|
def test_each_body_equals_the_span_its_own_entry_declares(tmp_path: Path) -> None:
|
|
source = drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
plan = build_plan(source.read_bytes(), DOCUMENT)
|
|
run(tmp_path, plan=plan)
|
|
found = concepts(tmp_path / "bundle")
|
|
for item in plan.entries:
|
|
start, end = item.span
|
|
assert body_of(found[item.path]) == DOCUMENT[start:end]
|
|
|
|
|
|
def test_concepts_land_on_nested_paths_across_several_directories(tmp_path: Path) -> None:
|
|
source = drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
run(tmp_path, plan=build_plan(source.read_bytes(), DOCUMENT))
|
|
found = concepts(tmp_path / "bundle")
|
|
directories = {str(Path(path).parent) for path in found}
|
|
assert len(directories) >= 2
|
|
nested = [path for path in found if "/" in path]
|
|
assert nested
|
|
# The concept ID is the bundle-relative path minus the suffix -- OKF v0.2
|
|
# §2's definition, not a name we assign.
|
|
for path in nested:
|
|
assert path.endswith(".md")
|
|
assert (tmp_path / "bundle" / path).is_file()
|
|
|
|
|
|
def test_the_plan_timestamp_reaches_every_concept(tmp_path: Path) -> None:
|
|
source = drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
run(tmp_path, plan=build_plan(source.read_bytes(), DOCUMENT))
|
|
for document in concepts(tmp_path / "bundle").values():
|
|
keys = frontmatter_of(document)
|
|
assert keys["ingested_at"] == PLAN_AT
|
|
assert keys["bundle_id"] == "b-1"
|
|
|
|
|
|
# --- S9: the span indexes EXTRACTED text, not source bytes ----------------
|
|
|
|
|
|
CSV_SOURCE = "krav;beskrivelse\n3-1;seksjonering\n3-2;roemning\n"
|
|
|
|
|
|
def test_a_csv_body_equals_the_span_of_the_extracted_text(tmp_path: Path) -> None:
|
|
# `render_table` re-renders these bytes, so extracted text != source bytes.
|
|
# A span computed against the bytes would land on different characters and
|
|
# produce a concept nobody adjudicated, with nothing failing.
|
|
source = drop(tmp_path / "round", "krav.csv", CSV_SOURCE)
|
|
text = extract_text("krav.csv", source.read_bytes())
|
|
assert text != CSV_SOURCE
|
|
|
|
plan = build_plan(
|
|
source.read_bytes(),
|
|
text,
|
|
paths=tuple(f"tabell/rad-{index}.md" for index in range(len(line_spans(text)))),
|
|
extractor_id="csv",
|
|
)
|
|
run(tmp_path, plan=plan)
|
|
found = concepts(tmp_path / "bundle")
|
|
assert len(found) == len(plan.entries)
|
|
for item in plan.entries:
|
|
start, end = item.span
|
|
assert body_of(found[item.path]) == text[start:end]
|
|
|
|
|
|
# --- fail-fast misuse, before any disk mutation ---------------------------
|
|
|
|
|
|
def test_a_plan_against_a_profile_without_the_capability_is_refused(tmp_path: Path) -> None:
|
|
source = drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
with pytest.raises(SegmentationError) as excinfo:
|
|
run(tmp_path, plan=build_plan(source.read_bytes(), DOCUMENT), profile=DEFAULT, values={})
|
|
assert excinfo.value.code == "segmentation_unsupported_profile"
|
|
assert tree(tmp_path / "bundle") == {}
|
|
|
|
|
|
def test_a_plan_without_a_bundle_id_is_refused_before_any_write(tmp_path: Path) -> None:
|
|
# The assertion Step 5 deferred to here: `process_inbox` gains its
|
|
# `segmentation` parameter at THIS step, so this is the first point at
|
|
# which the required-bundle_id branch can be reached at all.
|
|
source = drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
with pytest.raises(SegmentationError) as excinfo:
|
|
run(tmp_path, plan=build_plan(source.read_bytes(), DOCUMENT), values={})
|
|
assert excinfo.value.code == "segmentation_plan_invalid"
|
|
assert "bundle_id" in str(excinfo.value)
|
|
assert tree(tmp_path / "bundle") == {}
|
|
|
|
|
|
def test_a_refused_plan_leaves_an_existing_bundle_untouched(tmp_path: Path) -> None:
|
|
source = drop(tmp_path / "one", "n500.md", DOCUMENT)
|
|
run(tmp_path, plan=build_plan(source.read_bytes(), DOCUMENT), round_name="one")
|
|
before = tree(tmp_path / "bundle")
|
|
assert before != {}
|
|
|
|
drop(tmp_path / "two", "v720.md", DOCUMENT)
|
|
with pytest.raises(SegmentationError):
|
|
run(tmp_path, plan=build_plan(source.read_bytes(), DOCUMENT), values={}, round_name="two")
|
|
assert tree(tmp_path / "bundle") == before
|
|
|
|
|
|
# --- the whole-document refusal -------------------------------------------
|
|
|
|
|
|
def test_one_quarantined_segment_persists_nothing_for_that_document(tmp_path: Path) -> None:
|
|
source = drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
plan = build_plan(source.read_bytes(), DOCUMENT)
|
|
third = DOCUMENT[plan.entries[2].span[0] : plan.entries[2].span[1]]
|
|
|
|
def quarantining(text: str) -> GateDecision:
|
|
if text == third:
|
|
return GateDecision(
|
|
sanitized_text=text, disposition="quarantine_review", reasons=("segment 3",)
|
|
)
|
|
return GateDecision(sanitized_text=text, disposition="warn")
|
|
|
|
result = run(tmp_path, plan=plan, guard=quarantining)
|
|
assert concepts(tmp_path / "bundle") == {}
|
|
assert len(result.quarantined) == 1
|
|
assert result.quarantined[0].source_file == "n500.md"
|
|
assert result.persisted == ()
|
|
|
|
|
|
def test_every_segment_is_gated_before_any_is_written(tmp_path: Path) -> None:
|
|
source = drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
plan = build_plan(source.read_bytes(), DOCUMENT)
|
|
calls: list[str] = []
|
|
written_when_gated: list[int] = []
|
|
|
|
def recording(text: str) -> GateDecision:
|
|
calls.append(text)
|
|
written_when_gated.append(len(tree(tmp_path / "bundle")))
|
|
return GateDecision(sanitized_text=text, disposition="warn")
|
|
|
|
run(tmp_path, plan=plan, guard=recording)
|
|
assert len(calls) == 5
|
|
# Nothing on disk while any gate call is still outstanding.
|
|
assert written_when_gated == [0, 0, 0, 0, 0]
|
|
assert len(concepts(tmp_path / "bundle")) == 5
|
|
|
|
|
|
def test_a_plan_covering_one_of_two_documents_leaves_the_other_flat(tmp_path: Path) -> None:
|
|
# DIFFERENT bytes, deliberately. A plan is selected by content hash, so two
|
|
# files with identical content are both covered by one plan and land every
|
|
# segment on the same path -- which is the intra-run collision Step 8's gate
|
|
# exists to refuse, not something to demonstrate here.
|
|
source = drop(tmp_path / "round", "n500.md", DOCUMENT)
|
|
drop(tmp_path / "round", "v720.md", "V720 Tunnel: egne krav.\n")
|
|
run(tmp_path, plan=build_plan(source.read_bytes(), DOCUMENT))
|
|
found = concepts(tmp_path / "bundle")
|
|
assert set(PATHS) <= set(found)
|
|
assert "inbox-v720.md" in found
|
|
assert len(found) == 6
|