open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 1 Activity
llm-ingestion-okf/tests/test_segmentation.py

533 lines
20 KiB
Python
Raw Blame History

"""The segmentation plan: authored data that turns one document into many concepts.
A concept is "a single unit of knowledge within a bundle" (OKF v0.2 §2), not a
file that was dropped. Splitting one document into several is therefore a
judgement, and a judgement made by a model cannot live on a run path that
promises zero model calls. The resolution is to make the decision ONCE, write it
down as data, have a human adjudicate it, and replay it deterministically
thereafter — so this module is pure: no filesystem, no bundle, no door, no model.
Every path is normalised through the id grammar at entry, because macOS hands
filenames back DECOMPOSED: the same visual name would otherwise reduce two ways
depending on which form it arrived in, and silent ID motion between rounds is
the one failure this library cannot repair after the fact.
"""
from __future__ import annotations
import unicodedata
from typing import Any
import pytest
from llm_ingestion_okf.errors import SegmentationError
from llm_ingestion_okf.segmentation import (
SegmentationPlan,
SegmentEntry,
assert_plan_applies,
parse_segmentation_plan,
plan_cache_key,
slice_segments,
)
def entry(**overrides: Any) -> dict[str, Any]:
payload: dict[str, Any] = {
"segment_id": "s1",
"path": "krav/3-1/brannkonsept.md",
"title": "Brannkonsept",
"okf_type": "requirement",
"span": [0, 40],
"ingested_at": "2026-08-31T12:00:00Z",
}
payload.update(overrides)
return payload
def plan(**overrides: Any) -> dict[str, Any]:
payload: dict[str, Any] = {
"version": "1",
"source_sha256": "a" * 64,
"text_sha256": "c" * 64,
"extractor_id": "text",
"extractor_version": "1.0.0",
"adjudicated_at": "2026-08-31T11:00:00Z",
"entries": [entry()],
}
payload.update(overrides)
return payload
def parse_fails(payload: dict[str, Any]) -> SegmentationError:
with pytest.raises(SegmentationError) as excinfo:
parse_segmentation_plan(payload)
return excinfo.value
# --- the happy path -------------------------------------------------------
def test_a_valid_payload_parses_to_the_expected_plan() -> None:
parsed = parse_segmentation_plan(
plan(
entries=[
entry(),
entry(
segment_id="s2",
path="krav/3-1/roemning.md",
title="Roemning",
span=[40, 90],
parent_id="s1",
derived=["title"],
),
]
)
)
assert isinstance(parsed, SegmentationPlan)
assert parsed.version == "1"
assert parsed.source_sha256 == "a" * 64
assert parsed.extractor_id == "text"
assert parsed.extractor_version == "1.0.0"
assert parsed.adjudicated_at == "2026-08-31T11:00:00Z"
assert len(parsed.entries) == 2
first, second = parsed.entries
assert isinstance(first, SegmentEntry)
assert first.segment_id == "s1"
assert first.path == "krav/3-1/brannkonsept.md"
assert first.title == "Brannkonsept"
assert first.okf_type == "requirement"
assert first.span == (0, 40)
assert first.ingested_at == "2026-08-31T12:00:00Z"
assert first.parent_id is None
assert first.derived == frozenset()
assert second.parent_id == "s1"
assert second.derived == frozenset({"title"})
def test_entries_keep_plan_order_rather_than_being_sorted() -> None:
parsed = parse_segmentation_plan(
plan(
entries=[
entry(segment_id="s2", path="b.md", span=[10, 20]),
entry(segment_id="s1", path="a.md", span=[0, 10]),
]
)
)
assert [item.segment_id for item in parsed.entries] == ["s2", "s1"]
def test_the_plan_and_its_entries_are_frozen() -> None:
parsed = parse_segmentation_plan(plan())
with pytest.raises(Exception):
parsed.entries[0].path = "other.md" # type: ignore[misc]
with pytest.raises(Exception):
parsed.version = "2" # type: ignore[misc]
# --- the four codes this step's parser can raise --------------------------
#
# The registry names six. `segmentation_extractor_mismatch` is raised by
# `assert_plan_applies` and `segmentation_unsupported_profile` by Door B, so
# neither is reachable from the parser; each is covered where it becomes
# reachable rather than asserted here against code that does not exist yet.
def test_an_empty_entry_list_is_refused_as_plan_invalid() -> None:
error = parse_fails(plan(entries=[]))
assert error.code == "segmentation_plan_invalid"
def test_a_missing_top_level_field_is_refused_as_plan_invalid() -> None:
payload = plan()
del payload["extractor_version"]
error = parse_fails(payload)
assert error.code == "segmentation_plan_invalid"
assert "extractor_version" in str(error)
def test_a_missing_entry_field_is_refused_as_plan_invalid() -> None:
broken = entry()
del broken["okf_type"]
error = parse_fails(plan(entries=[broken]))
assert error.code == "segmentation_plan_invalid"
assert "okf_type" in str(error)
def test_a_parent_id_naming_no_entry_is_refused_as_plan_invalid() -> None:
error = parse_fails(plan(entries=[entry(parent_id="nobody")]))
assert error.code == "segmentation_plan_invalid"
assert "nobody" in str(error)
def test_a_leading_slash_is_refused_as_path_invalid() -> None:
error = parse_fails(plan(entries=[entry(path="/krav/a.md")]))
assert error.code == "segmentation_path_invalid"
def test_a_parent_traversal_component_is_refused_as_path_invalid() -> None:
error = parse_fails(plan(entries=[entry(path="krav/../../etc/passwd.md")]))
assert error.code == "segmentation_path_invalid"
def test_a_component_reducing_to_nothing_is_refused_as_path_invalid() -> None:
error = parse_fails(plan(entries=[entry(path="krav/---/a.md")]))
assert error.code == "segmentation_path_invalid"
def test_two_entries_on_one_path_are_refused_as_path_invalid() -> None:
error = parse_fails(
plan(
entries=[
entry(segment_id="s1", path="krav/a.md", span=[0, 10]),
entry(segment_id="s2", path="krav/a.md", span=[10, 20]),
]
)
)
assert error.code == "segmentation_path_invalid"
assert "krav/a.md" in str(error)
def test_two_entries_sharing_an_id_are_refused_as_duplicate_id() -> None:
error = parse_fails(
plan(
entries=[
entry(segment_id="s1", path="krav/a.md", span=[0, 10]),
entry(segment_id="s1", path="krav/b.md", span=[10, 20]),
]
)
)
assert error.code == "segmentation_duplicate_id"
assert "s1" in str(error)
def test_a_negative_span_offset_is_refused_as_span_invalid() -> None:
error = parse_fails(plan(entries=[entry(span=[-1, 10])]))
assert error.code == "segmentation_span_invalid"
def test_a_span_that_does_not_advance_is_refused_as_span_invalid() -> None:
error = parse_fails(plan(entries=[entry(span=[10, 10])]))
assert error.code == "segmentation_span_invalid"
def test_a_reversed_span_is_refused_as_span_invalid() -> None:
error = parse_fails(plan(entries=[entry(span=[40, 10])]))
assert error.code == "segmentation_span_invalid"
def test_a_span_of_the_wrong_shape_is_refused_as_span_invalid() -> None:
error = parse_fails(plan(entries=[entry(span=[0, 10, 20])]))
assert error.code == "segmentation_span_invalid"
# --- normalisation --------------------------------------------------------
def test_a_decomposed_path_normalises_to_the_composed_one() -> None:
# macOS/APFS hands names back decomposed. The same visual path must reduce
# to the same stored path from either form, or a re-run silently moves an
# ID -- the failure this library cannot repair afterwards.
composed = unicodedata.normalize("NFC", "krav/bygningsdelér.md")
decomposed = unicodedata.normalize("NFD", composed)
assert composed != decomposed
from_composed = parse_segmentation_plan(plan(entries=[entry(path=composed)]))
from_decomposed = parse_segmentation_plan(plan(entries=[entry(path=decomposed)]))
assert from_composed.entries[0].path == from_decomposed.entries[0].path
stored = from_composed.entries[0].path
assert stored == unicodedata.normalize("NFC", stored)
def test_a_path_is_lowercased_and_reduced_component_by_component() -> None:
parsed = parse_segmentation_plan(plan(entries=[entry(path="Krav/3.1 Brann/Konsept A.md")]))
assert parsed.entries[0].path == "krav/3-1-brann/konsept-a.md"
def test_the_final_suffix_survives_reduction() -> None:
parsed = parse_segmentation_plan(plan(entries=[entry(path="krav/brannkonsept.md")]))
assert parsed.entries[0].path.endswith(".md")
def test_two_paths_differing_only_in_normal_form_collide_as_duplicates() -> None:
composed = unicodedata.normalize("NFC", "krav/bygningsdelér.md")
decomposed = unicodedata.normalize("NFD", composed)
error = parse_fails(
plan(
entries=[
entry(segment_id="s1", path=composed, span=[0, 10]),
entry(segment_id="s2", path=decomposed, span=[10, 20]),
]
)
)
assert error.code == "segmentation_path_invalid"
# --- S5b: the cache key is the extractor, not the hash alone ---------------
#
# Source bytes cannot see an extractor swap or a version bump. Both invalidate
# every stored offset while `source_sha256` stays identical, so the mismatch
# has to be LOUD -- a silent re-derivation would replay a human's adjudication
# against text that human never saw.
def parsed_plan(**overrides: Any) -> SegmentationPlan:
return parse_segmentation_plan(plan(**overrides))
def applies_fails(subject: SegmentationPlan, **overrides: str) -> SegmentationError:
arguments = {
"source_sha256": subject.source_sha256,
"text_sha256": subject.text_sha256,
"extractor_id": subject.extractor_id,
"extractor_version": subject.extractor_version,
}
arguments.update(overrides)
with pytest.raises(SegmentationError) as excinfo:
assert_plan_applies(subject, **arguments)
return excinfo.value
def test_the_cache_key_is_the_four_tuple() -> None:
subject = parsed_plan()
assert plan_cache_key(subject) == (
subject.source_sha256,
subject.text_sha256,
subject.extractor_id,
subject.extractor_version,
)
def test_two_plans_differing_only_in_extractor_id_have_different_cache_keys() -> None:
# The whole point of S5b: the hash alone would call these one cached
# adjudication, and replay the first plan's offsets against the second
# extraction.
one = parsed_plan(extractor_id="text")
other = parsed_plan(extractor_id="pdfplumber")
assert one.source_sha256 == other.source_sha256
assert plan_cache_key(one) != plan_cache_key(other)
assert plan_cache_key(one)[0] == plan_cache_key(other)[0]
def test_an_identical_quadruple_applies_without_raising() -> None:
subject = parsed_plan()
assert (
assert_plan_applies(
subject,
source_sha256=subject.source_sha256,
text_sha256=subject.text_sha256,
extractor_id=subject.extractor_id,
extractor_version=subject.extractor_version,
)
is None
)
def test_a_changed_text_hash_is_refused_although_the_source_bytes_match() -> None:
"""The component the other three cannot stand in for.
Same bytes, same extractor, same version -- and a canonical text that
moved anyway, which is what a converter reshaping its output without
bumping its version looks like from here. Before this component existed
the key matched and every offset was replayed against text nobody
adjudicated, with each span still landing on real characters.
"""
error = applies_fails(parsed_plan(), text_sha256="d" * 64)
assert error.code == "segmentation_extractor_mismatch"
assert "text_sha256" in str(error)
assert "source_sha256" not in str(error)
def test_a_changed_extractor_version_is_refused_and_named() -> None:
error = applies_fails(parsed_plan(), extractor_version="1.0.1")
assert error.code == "segmentation_extractor_mismatch"
assert "extractor_version" in str(error)
assert "1.0.1" in str(error)
def test_a_changed_extractor_id_is_refused_and_named() -> None:
error = applies_fails(parsed_plan(), extractor_id="pdfplumber")
assert error.code == "segmentation_extractor_mismatch"
assert "extractor_id" in str(error)
def test_a_changed_source_hash_is_refused_and_named() -> None:
error = applies_fails(parsed_plan(), source_sha256="b" * 64)
assert error.code == "segmentation_extractor_mismatch"
assert "source_sha256" in str(error)
def test_the_message_names_every_differing_component_not_just_the_first() -> None:
error = applies_fails(parsed_plan(), extractor_id="pdfplumber", extractor_version="1.0.1")
assert "extractor_id" in str(error)
assert "extractor_version" in str(error)
assert "source_sha256" not in str(error)
# --- slicing: spans are windows on the EXTRACTED text ----------------------
#
# Not on the source bytes. You cannot slice a PDF's bytes and recover prose,
# and a `.csv` is re-rendered as a table before it becomes a concept body --
# so an offset computed against bytes would land on different characters and
# quietly produce a concept nobody adjudicated.
SECTIONS = (
"0 Forord: bakgrunn for konseptet.\n" # 0..34
"1 Brannkonsept: krav til seksjonering.\n" # 34..73
"2 Roemning: to uavhengige veier.\n" # 73..106
"3 Baereevne: R60 for hovedbaeresystem.\n" # 106..145
"4 Slokkeanlegg: sprinkler i hele bygget.\n" # 145..186
)
SPANS = ((0, 34), (34, 73), (73, 106), (106, 145), (145, 186))
def five_section_plan(**overrides: Any) -> SegmentationPlan:
spans = overrides.pop("spans", SPANS)
return parse_segmentation_plan(
plan(
entries=[
entry(segment_id=f"s{index}", path=f"krav/{index}-del.md", span=list(span))
for index, span in enumerate(spans)
],
**overrides,
)
)
def test_five_declared_spans_yield_those_five_substrings_in_plan_order() -> None:
sliced = slice_segments(SECTIONS, five_section_plan())
assert len(sliced) == 5
assert [item.segment_id for item, _ in sliced] == ["s0", "s1", "s2", "s3", "s4"]
assert [body for _, body in sliced] == [SECTIONS[start:end] for start, end in SPANS]
assert sliced[1][1].startswith("1 Brannkonsept")
def test_moving_one_span_changes_exactly_that_one_body() -> None:
# The S1 correspondence anchor: a concept's body is the span its OWN entry
# declares. If a neighbour's body moved too, the split would be an artefact
# of iteration rather than of the adjudication.
def bodies_by_id(subject: SegmentationPlan) -> dict[str, str]:
return {item.segment_id: body for item, body in slice_segments(SECTIONS, subject)}
baseline = bodies_by_id(five_section_plan())
moved_spans = list(SPANS)
moved_spans[2] = (73, 90)
moved = bodies_by_id(five_section_plan(spans=tuple(moved_spans)))
assert set(moved) == set(baseline)
assert {key for key in moved if moved[key] != baseline[key]} == {"s2"}
def test_a_span_reaching_past_the_end_is_refused() -> None:
subject = five_section_plan(spans=((0, 34), (34, len(SECTIONS) + 1)))
with pytest.raises(SegmentationError) as excinfo:
slice_segments(SECTIONS, subject)
assert excinfo.value.code == "segmentation_span_invalid"
assert str(len(SECTIONS)) in str(excinfo.value)
def test_a_span_ending_exactly_at_the_end_is_permitted() -> None:
sliced = slice_segments(SECTIONS, five_section_plan(spans=((145, len(SECTIONS)),)))
assert sliced[0][1] == SECTIONS[145:]
def test_an_uncovered_gap_in_the_middle_is_permitted() -> None:
# A preamble, a page header or a signature block may belong to no concept.
# Refusing a gap would force the adjudicator to invent a home for text that
# is not a unit of knowledge.
sliced = slice_segments(SECTIONS, five_section_plan(spans=((0, 34), (106, 145))))
assert [body for _, body in sliced] == [SECTIONS[0:34], SECTIONS[106:145]]
def test_overlapping_spans_are_permitted() -> None:
sliced = slice_segments(SECTIONS, five_section_plan(spans=((0, 73), (34, 106))))
assert sliced[0][1] == SECTIONS[0:73]
assert sliced[1][1] == SECTIONS[34:106]
def test_slicing_an_empty_text_with_any_span_is_refused() -> None:
with pytest.raises(SegmentationError) as excinfo:
slice_segments("", five_section_plan(spans=((0, 1),)))
assert excinfo.value.code == "segmentation_span_invalid"
# --- S5c: the quote anchor, which survives a shift the key must not refuse --
#
# The key (S5b, above) refuses a FOREIGN extraction. This is the other half:
# a benign shift INSIDE the same extraction, where the text an entry names is
# still present but no longer at the offset written down. Measured: changing
# one extraction flag moved a document from 15 507 to 8 290 characters and
# every inspected span pointed at the wrong text. Offsets are therefore a HINT
# the quote may correct -- never the last word, and never silently wrong.
ANCHORED = "innledning\n" + SECTIONS
def anchored_entry(index: int, **overrides: Any) -> dict[str, Any]:
start, end = SPANS[index]
quote = SECTIONS[start:end]
payload = entry(
segment_id=f"s{index}",
path=f"krav/{index}-del.md",
span=[start, end],
anchor={
"quote": quote,
"prefix": SECTIONS[max(0, start - 12) : start],
"suffix": SECTIONS[end : end + 12],
},
)
payload.update(overrides)
return payload
def anchored_plan(*indexes: int, **overrides: Any) -> SegmentationPlan:
return parse_segmentation_plan(
plan(entries=[anchored_entry(index) for index in indexes], **overrides)
)
def test_an_anchor_whose_offsets_still_hold_changes_nothing() -> None:
"""The fast path. A correct plan must not be re-anchored into motion."""
sliced = slice_segments(SECTIONS, anchored_plan(1, 2))
assert [body for _, body in sliced] == [SECTIONS[34:73], SECTIONS[73:106]]
assert [item.span for item, _ in sliced] == [(34, 73), (73, 106)]
def test_stale_offsets_re_anchor_to_the_text_the_quote_names() -> None:
"""The whole point: the same document with eleven characters prepended.
Every declared offset is now short by exactly that much. Without the
anchor each body would be cut eleven characters early and land on real
text -- the failure no downstream assertion can see.
"""
sliced = slice_segments(ANCHORED, anchored_plan(1, 2))
assert [body for _, body in sliced] == [SECTIONS[34:73], SECTIONS[73:106]]
assert [item.span for item, _ in sliced] == [(45, 84), (84, 117)]
def test_a_quote_that_no_longer_occurs_is_refused_rather_than_cut_elsewhere() -> None:
"""Refusing beats cutting. A body sliced at a plausible-but-wrong offset
reads as ordinary prose, so nothing downstream can distinguish it from an
adjudicated one."""
gone = SECTIONS.replace("1 Brannkonsept: krav til seksjonering.\n", "")
with pytest.raises(SegmentationError) as excinfo:
slice_segments(gone, anchored_plan(1))
assert excinfo.value.code == "segmentation_span_invalid"
assert "s1" in str(excinfo.value)
def test_a_repeated_quote_re_anchors_to_the_occurrence_its_context_names() -> None:
"""Prefix and suffix are what make a repeated line addressable at all."""
doubled = "xx" + SECTIONS + SECTIONS
sliced = slice_segments(doubled, anchored_plan(1))
assert sliced[0][0].span == (36, 75)
assert sliced[0][1] == SECTIONS[34:73]
def test_an_entry_without_an_anchor_keeps_todays_offset_behaviour() -> None:
"""The anchor is a capability, not a new requirement on authored plans."""
sliced = slice_segments(SECTIONS, five_section_plan())
assert [item.span for item, _ in sliced] == list(SPANS)
Reference in a new issue View git blame Copy permalink