llm-ingestion-okf/tests/test_provenance.py

"""Provenance to the original: `sources` plus a per-format locator (O3).

A concept says what it was extracted FROM. Before this step it said so with a
`source_file` basename, a `source_sha256`, and — when segmented — a
`source_offset` that indexes the EXTRACTED text rather than the original, so a
consumer could not open the original at the right place without knowing the
corpus directory and re-running the extractor.

Two layers, and the split is the whole design:

- the ADDRESS is spec's, `sources[].resource` (SPEC v0.2 §5.1:303-306: "an
  absolute URL, a bundle-relative path, or a path into a `references/`
  subdirectory"), emitted in flow form because this library's parser cannot
  read a block one back;
- the LOCATOR is ours, because §5.1 has no field for a page, a sheet row or a
  line, and the guard's frontmatter grammar refuses both a non-allowlisted key
  inside a `sources` entry and a nested flow list — measured here, so the
  choice is a recorded constraint rather than a preference.
"""

from __future__ import annotations

import importlib.util
import warnings
from pathlib import Path

import pytest

from llm_ingestion_okf.extract import extract_text, source_units
from llm_ingestion_okf.inbox import render_inbox_concept
from llm_ingestion_okf.profiles import (
    DEFAULT,
    SEGMENTED_OKF_V0_2,
    SEGMENTED_V1,
    STRICT_V1,
    STRUCTURED_V1,
)
from llm_ingestion_okf.propose import RULE_SHEET_SECTION, find_candidates
from llm_ingestion_okf.segmentation import SegmentEntry

FIXTURES = Path(__file__).parent / "fixtures"

requires_extract = pytest.mark.skipif(
    importlib.util.find_spec("pdfplumber") is None,
    reason="the optional [extract] extra is not installed",
)


def _extract(name: str) -> tuple[bytes, str]:
    data = (FIXTURES / name).read_bytes()
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        return data, extract_text(name, data)


def _segment(span: tuple[int, int]) -> SegmentEntry:
    return SegmentEntry(
        segment_id="p1",
        path="a/b.md",
        title="A segment",
        okf_type="reference",
        span=span,
        ingested_at="2026-09-08T00:00:00Z",
    )


# --- the unit table: a character range -> a place in the ORIGINAL ---------


@requires_extract
def test_a_pdf_reports_the_page_a_character_range_came_from() -> None:
    # The middle page carries no text and the extractor drops it, so the third
    # page's text is page 3 and not page 2. A fixture without that gap could
    # not tell a page NUMBER from a count of the pages that produced text.
    data, text = _extract("three-page-krav.pdf")
    units = source_units("three-page-krav.pdf", data, text)
    assert units is not None
    assert units.unit == "pages"
    assert units.covering(0, len("Side en om helning")) == (1, 1)
    assert units.covering(text.index("Side tre"), len(text)) == (3, 3)
    assert units.covering(0, len(text)) == (1, 3)
    # The BOUNDARY, and it is the assertion that has to exist: the blank line
    # the extractor joins pages with belongs to neither page's text, so an
    # offset inside it is still the page before. A table built without the
    # separator's own length passes every assertion above and fails this one,
    # because the drift is two characters per page and only shows up where a
    # page begins.
    last_page_begins = text.index("Side tre")
    assert units.covering(last_page_begins - 1, last_page_begins) == (1, 1)


@requires_extract
def test_a_spreadsheet_reports_the_sheet_and_the_rows_of_the_original() -> None:
    # `prisark.xlsx` is hand-laid: sheet 1 spans A1:C6, sheet 2 spans A1:A3.
    # The converter renders each sheet as a heading plus a pipe table, and
    # numbering restarts per sheet — so the last row of the file is row 3 of
    # sheet 2, not row 9 of the document.
    data, text = _extract("prisark.xlsx")
    units = source_units("prisark.xlsx", data, text)
    assert units is not None
    assert units.unit == "rows"
    first_row = text.index("| Prisskjema")
    assert units.covering(first_row, first_row + 1) == (1, 1)
    assert units.scope_of(first_row) == "Prisark"
    assert units.covering(0, len(text)) == (1, 3)
    assert units.scopes_covering(0, len(text)) == ("Prisark", "Enkeltkolonne")


@requires_extract
def test_a_row_locator_counts_the_separator_line_as_no_row() -> None:
    # The pipe table's `|----|` line is a row of no spreadsheet. Counting it
    # would push every row after it up by one, silently, for the whole sheet.
    data, text = _extract("prisark.xlsx")
    units = source_units("prisark.xlsx", data, text)
    assert units is not None
    header = text.index("| Prisskjema")
    second = text.index("| Post ")
    assert units.covering(header, second + 1) == (1, 2)


def test_an_empty_row_still_counts_as_a_row_of_the_sheet() -> None:
    # `tomrad.xlsx` is four rows with the THIRD empty. The converter renders an
    # empty row as a pipe line of nothing but spaces, which is what a table's
    # own separator line also looks like — so a rule that reads the LINE rather
    # than its POSITION swallows the empty row and renumbers every row after
    # it, for the whole sheet, silently.
    #
    # Found on the K2 price sheet, not here: 8 empty rows, and the last row
    # reported as 92 against a workbook that says 100.
    data, text = _extract("tomrad.xlsx")
    units = source_units("tomrad.xlsx", data, text)
    assert units is not None
    assert units.numbers == (1, 2, 3, 4)
    assert units.covering(text.index("Rad fire"), len(text)) == (4, 4)


def test_a_text_file_reports_line_numbers_of_the_original() -> None:
    data = b"first\nsecond\nthird\n"
    text = extract_text("note.md", data)
    units = source_units("note.md", data, text)
    assert units is not None
    assert units.unit == "lines"
    assert units.covering(0, 5) == (1, 1)
    assert units.covering(text.index("third"), len(text)) == (3, 3)
    assert units.covering(0, len(text)) == (1, 3)


def test_a_docx_locator_is_lines_because_paragraphs_do_not_survive() -> None:
    # Measured on the five K2 `.docx` documents: `<w:p>` counts of 108, 27, 65,
    # 176 and 57 against converted-markdown line counts of 75, 33, 67, 144 and
    # 63. Not one pair agrees, so a `paragraphs` key would name a number the
    # original does not have. `lines` says what it indexes.
    data, text = _extract("two-line-krav.docx") if _has_converter() else (b"", "")
    if not text:
        pytest.skip("the optional [extract] extra is not installed")
    units = source_units("two-line-krav.docx", data, text)
    assert units is not None
    assert units.unit == "lines"


def _has_converter() -> bool:
    return importlib.util.find_spec("pypandoc") is not None


# --- the frontmatter a concept carries ------------------------------------


@requires_extract
def test_a_segmented_concept_points_at_the_original_and_its_pages() -> None:
    data, text = _extract("three-page-krav.pdf")
    units = source_units("three-page-krav.pdf", data, text)
    document = render_inbox_concept(
        text[20:],
        okf_type="reference",
        title="Side tre",
        source_file="mappe/three-page-krav.pdf",
        source_bytes=data,
        ingested_at="2026-09-08T00:00:00Z",
        profile=SEGMENTED_OKF_V0_2,
        segment=_segment((20, len(text))),
        bundle_id="b1",
        units=units,
    )
    assert "sources: [{ resource: mappe/three-page-krav.pdf, title: three-page-krav.pdf }]\n" in (
        document
    )
    assert "source_pages: [3, 3]\n" in document
    # The offset stays: it is what an existing consumer joins on, and a
    # locator that replaced it would break them to fix them.
    assert "source_offset: [20, 40]\n" in document


@requires_extract
def test_a_whole_document_concept_gets_the_pages_it_spans() -> None:
    # The order's known-negative: a concept the plan does not cover is the
    # WHOLE document, so its locator is every page that produced text.
    data, text = _extract("three-page-krav.pdf")
    units = source_units("three-page-krav.pdf", data, text)
    document = render_inbox_concept(
        text,
        okf_type="reference",
        title="Hele",
        source_file="three-page-krav.pdf",
        source_bytes=data,
        ingested_at="2026-09-08T00:00:00Z",
        profile=SEGMENTED_OKF_V0_2,
        units=units,
        span=(0, len(text)),
    )
    assert "source_pages: [1, 3]\n" in document
    assert "source_offset" not in document


@requires_extract
def test_a_spreadsheet_concept_names_the_sheet_and_its_rows() -> None:
    data, text = _extract("prisark.xlsx")
    units = source_units("prisark.xlsx", data, text)
    document = render_inbox_concept(
        text,
        okf_type="reference",
        title="Hele arket",
        source_file="prisark.xlsx",
        source_bytes=data,
        ingested_at="2026-09-08T00:00:00Z",
        profile=SEGMENTED_OKF_V0_2,
        units=units,
        span=(0, text.index("| Post ")),
    )
    assert "source_sheet: Prisark\n" in document
    assert "source_rows: [1, 1]\n" in document


@requires_extract
def test_each_sheet_section_names_its_own_rows() -> None:
    """D3: a section's locator is the section's rows, not the sheet's.

    The rule cuts one table block into one candidate per numbered row, so the
    provenance layer is what decides whether those concepts are addressable at
    all. It reads the unit table built AT EXTRACTION and the candidate's own
    span, and neither knows about the rule — which is exactly why this has to
    be measured rather than assumed: a locator that reported the whole sheet
    for every section would look right in the frontmatter and point at nothing.
    """
    data, text = _extract("prisark.xlsx")
    units = source_units("prisark.xlsx", data, text)
    candidates = find_candidates(text, sheet_section_rows=True)
    sections = [c for c in candidates if c.rule == RULE_SHEET_SECTION]
    assert [c.number for c in sections] == ["01", "02", "03", "04"]

    located = []
    for candidate in sections:
        document = render_inbox_concept(
            text,
            okf_type="reference",
            title=candidate.title,
            source_file="prisark.xlsx",
            source_bytes=data,
            ingested_at="2026-09-08T00:00:00Z",
            profile=SEGMENTED_OKF_V0_2,
            units=units,
            span=(candidate.start, candidate.end),
        )
        assert "source_sheet: Prisark\n" in document
        located.append(
            next(line for line in document.splitlines() if line.startswith("source_rows:"))
        )
    # Four sections, four DIFFERENT row locators, ascending, and each one is
    # the workbook's own row number rather than a line in the extracted text.
    # The last section stops at the second sheet's heading, which is the next
    # mark: a span crossing two sheets would get no sheet and no rows at all,
    # and the fixture is what proves it does not.
    assert located == [
        "source_rows: [3, 3]",
        "source_rows: [4, 4]",
        "source_rows: [5, 5]",
        "source_rows: [6, 6]",
    ]


@requires_extract
def test_a_range_spanning_two_sheets_names_no_sheet_and_no_rows() -> None:
    # A row number is only a place in the original once a sheet is named. A
    # range covering two sheets has no single sheet, so it gets no row
    # locator either — an absence, never a first-sheet guess.
    data, text = _extract("prisark.xlsx")
    units = source_units("prisark.xlsx", data, text)
    document = render_inbox_concept(
        text,
        okf_type="reference",
        title="Begge ark",
        source_file="prisark.xlsx",
        source_bytes=data,
        ingested_at="2026-09-08T00:00:00Z",
        profile=SEGMENTED_OKF_V0_2,
        units=units,
        span=(0, len(text)),
    )
    assert "source_sheet" not in document
    assert "source_rows" not in document
    assert "sources: [{ resource: prisark.xlsx, title: prisark.xlsx }]\n" in document


def test_the_address_is_written_even_when_no_locator_is_available() -> None:
    # `sources` answers "which file", the locator answers "where in it". The
    # first must not depend on the second: a caller with no unit table still
    # owes a consumer the address.
    document = render_inbox_concept(
        "body\n",
        okf_type="reference",
        title="T",
        source_file="sub/dir/note.md",
        source_bytes=b"body\n",
        ingested_at="2026-09-08T00:00:00Z",
        profile=SEGMENTED_OKF_V0_2,
    )
    assert "sources: [{ resource: sub/dir/note.md, title: note.md }]\n" in document
    assert "source_pages" not in document
    assert "source_lines" not in document


# --- what must not move ----------------------------------------------------


@pytest.mark.parametrize("profile", [DEFAULT, STRUCTURED_V1, STRICT_V1, SEGMENTED_V1])
def test_a_profile_that_names_no_provenance_writes_none(profile: object) -> None:
    # Support is additive: five shipped profiles keep their bytes, and only the
    # profile the order targets moves. A key written unconditionally here would
    # churn every golden in the suite.
    assert getattr(profile, "provenance") is None


def test_the_shipped_profiles_that_move_are_exactly_one() -> None:
    assert SEGMENTED_OKF_V0_2.provenance is not None
    for profile in (DEFAULT, STRUCTURED_V1, STRICT_V1, SEGMENTED_V1):
        assert profile.provenance is None


def test_a_source_file_that_would_break_the_flow_mapping_is_refused() -> None:
    # Validation, not repair, and not silence: the emitted value is a YAML flow
    # mapping, so a comma or a brace in the path would terminate the entry
    # early and produce a `sources` list that parses as something else.
    from llm_ingestion_okf.errors import MaterializationError

    with pytest.raises(MaterializationError) as excinfo:
        render_inbox_concept(
            "body\n",
            okf_type="reference",
            title="T",
            source_file="Del II, Bilag.pdf",
            source_bytes=b"body\n",
            ingested_at="2026-09-08T00:00:00Z",
            profile=SEGMENTED_OKF_V0_2,
        )
    assert excinfo.value.code == "inbox_source_file_unaddressable"


def test_the_guard_parses_the_sources_form_this_door_emits() -> None:
    # The published promise this test exists to keep red-able: what Door B
    # writes must survive the guard's own frontmatter grammar, or a bundle we
    # emit could never be read back through Door C.
    okf = pytest.importorskip("llm_ingestion_guard.okf")
    document = render_inbox_concept(
        "body\n",
        okf_type="reference",
        title="T",
        source_file="Del II Bilag 3.3.1 - Brannkonsept.pdf",
        source_bytes=b"body\n",
        ingested_at="2026-09-08T00:00:00Z",
        profile=SEGMENTED_OKF_V0_2,
    )
    frontmatter, _ = okf.parse_frontmatter(document)
    assert frontmatter["sources"] == [
        {
            "resource": "Del II Bilag 3.3.1 - Brannkonsept.pdf",
            "title": "Del II Bilag 3.3.1 - Brannkonsept.pdf",
        }
    ]