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

434 lines
16 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,
"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,
"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_three_tuple() -> None:
subject = parsed_plan()
assert plan_cache_key(subject) == (
subject.source_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_triple_applies_without_raising() -> None:
subject = parsed_plan()
assert (
assert_plan_applies(
subject,
source_sha256=subject.source_sha256,
extractor_id=subject.extractor_id,
extractor_version=subject.extractor_version,
)
is None
)
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"
Reference in a new issue View git blame Copy permalink