llm-ingestion-okf/tests/test_block_sources_flat_readers.py

"""A block `sources:` sequence must reach EVERY flat reader (K3-24).

`consume.read_sources` reads both YAML forms; this library's three copies of
the line-oriented frontmatter grammar read only the flow one. On a block
sequence they return the key with an EMPTY value -- not a `KeyError` a
consumer can catch, an address that disappears. Measured 2026-09-12 over four
bundles a producer ships: 2 756 of 2 757, 446 of 447, 1 133 of 1 134 and 270
of 271 concept files lost the address that way, while PyYAML 6.0.3 and the
pinned guard 1.4.0 both read it.

The bar is dict EQUALITY against an independent reader, not "it parses":
a reader that returned something plausible would satisfy the weaker claim.
PyYAML and the guard are compared separately because they disagree on ONE
axis and the disagreement is a fact rather than a defect -- the guard keeps a
leaf's quotes verbatim, PyYAML decodes them. This library follows its own
K3-22 rule (`unquote_scalar`: a `"`-wrapped leaf is decoded, a `'`-wrapped one
stands), which is `read_sources`' rule, so the reference for a QUOTED leaf is
`read_sources` and the reference for a plain one is all three at once.
"""

from pathlib import Path

import pytest
import yaml
from llm_ingestion_guard import okf as guard_okf

from llm_ingestion_okf.consume import _frontmatter_lines, _parse_flow_mappings, read_sources
from llm_ingestion_okf.materialize import parse_frontmatter
from llm_ingestion_okf.profiles import STRUCTURED_BLOCK_KEYS
from llm_ingestion_okf.profiles import _split_frontmatter as _profiles_split
from llm_ingestion_okf.structure import _split_frontmatter as _structure_split

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

# The form the producer actually ships, copied from a bundle: a query string
# carries `?`, `=` and `&`, and no flow form of it passes both PyYAML and the
# guard -- which is why the producer moved to the block form at all.
VEGNORMAL = """\
---
type: Krav
title: Krav 10.2-2 Beredskap
source_file: normal.xml
sources:
  - resource: https://viewers.test.invalid/api/nisosts/860019?languageCode=nb&v=2
    title: N500:2024
---

## Krav

Body.
"""

# Exactly the case the README says collapses into one namespace with the first
# entry lost. The test measures what happens instead of repeating the claim.
TWO_ENTRIES = """\
---
title: Krav 1
sources:
  - resource: a.pdf
    title: A
  - resource: b.pdf
    title: B
---

Body.
"""

# K3-20's substitution trap: the entry carries a `title:` of its own, and the
# document already has one. Reading the block may never let a nested key into
# the document's namespace.
NESTED_TITLE = """\
---
title: N100.2 Kryss og avkjoersler
generated: true
source_file: vegnormal.md
sources:
  - resource: https://example.test/bruprosjektering.pdf
    title: N200.7 Bruprosjektering
---

Body.
"""

FLOW = """\
---
title: Krav 1
generated: { by: process:okf-ingest, at: 2026-08-31T00:00:00Z }
sources: [{ resource: a.pdf, title: A }]
---

Body.
"""

NO_SOURCES = """\
---
title: Krav 1
source_file: normal.md
---

Body.
"""

# `read_sources` refuses this (consume.py: "An indented line before any `- `
# opens no entry"). The flat readers must reach the same conclusion instead of
# inventing an entry the document does not have.
INDENTED_BEFORE_DASH = """\
---
title: Krav 1
sources:
    resource: a.pdf
---

Body.
"""

QUOTED_LEAVES = """\
---
title: Krav 1
sources:
  - resource: "a, b.pdf"
    title: 'N100'
---

Body.
"""


def _flat_readings(text: str, tmp_path: Path) -> dict[str, dict[str, str]]:
    """The same document through all three copies of the flat grammar."""
    path = tmp_path / "concept.md"
    path.write_text(text, encoding="utf-8")
    return {
        "materialize": parse_frontmatter(path),
        "structure": _structure_split(text)[0],
        "profiles": _profiles_split(text)[0],
    }


def _pyyaml_sources(text: str) -> object:
    block = text.split("---\n")[1]
    loaded = yaml.safe_load(block)
    assert isinstance(loaded, dict)
    return loaded.get("sources")


def _guard_sources(text: str) -> object:
    return guard_okf.parse_frontmatter(text)[0].get("sources")


def _read_sources(text: str, tmp_path: Path) -> tuple[tuple[object, ...], bool]:
    path = tmp_path / "for-read-sources.md"
    path.write_text(text, encoding="utf-8")
    return read_sources(_frontmatter_lines(path))


# --- known-positive controls ---------------------------------------------
#
# Without these, a reader that decoded nothing at all would satisfy several
# assertions below. Each control proves the SAME reader finds the address in
# the form it already supports.


def test_control_read_sources_reads_the_block_form(tmp_path: Path) -> None:
    entries, present = _read_sources(VEGNORMAL, tmp_path)
    assert present
    assert [dict(entry) for entry in entries] == [
        {
            "resource": "https://viewers.test.invalid/api/nisosts/860019?languageCode=nb&v=2",
            "title": "N500:2024",
        }
    ]


def test_control_both_reference_readers_read_the_block_form() -> None:
    expected = [
        {
            "resource": "https://viewers.test.invalid/api/nisosts/860019?languageCode=nb&v=2",
            "title": "N500:2024",
        }
    ]
    assert _pyyaml_sources(VEGNORMAL) == expected
    assert _guard_sources(VEGNORMAL) == expected


# --- the defect, once per copy of the grammar -----------------------------


@pytest.mark.parametrize("reader", ["materialize", "structure", "profiles"])
def test_block_sources_reaches_every_flat_reader(reader: str, tmp_path: Path) -> None:
    value = _flat_readings(VEGNORMAL, tmp_path)[reader]["sources"]
    assert value != ""
    assert _parse_flow_mappings(value) == _pyyaml_sources(VEGNORMAL)
    assert _parse_flow_mappings(value) == _guard_sources(VEGNORMAL)


@pytest.mark.parametrize("reader", ["materialize", "structure", "profiles"])
def test_two_entries_both_survive(reader: str, tmp_path: Path) -> None:
    """The README's own example: two mappings, neither lost."""
    value = _flat_readings(TWO_ENTRIES, tmp_path)[reader]["sources"]
    assert _parse_flow_mappings(value) == [
        {"resource": "a.pdf", "title": "A"},
        {"resource": "b.pdf", "title": "B"},
    ]
    assert _parse_flow_mappings(value) == _pyyaml_sources(TWO_ENTRIES)
    assert _parse_flow_mappings(value) == _guard_sources(TWO_ENTRIES)


# --- the known-negatives the fix may not move -----------------------------


@pytest.mark.parametrize("reader", ["materialize", "structure", "profiles"])
def test_a_nested_title_still_does_not_substitute(reader: str, tmp_path: Path) -> None:
    """The most important control here: reading the block is not a licence to
    let a nested key into the document's namespace."""
    flat = _flat_readings(NESTED_TITLE, tmp_path)[reader]
    assert flat["title"] == "N100.2 Kryss og avkjoersler"
    assert "resource" not in flat
    assert set(flat) == {"title", "generated", "source_file", "sources"}
    assert _parse_flow_mappings(flat["sources"]) == [
        {
            "resource": "https://example.test/bruprosjektering.pdf",
            "title": "N200.7 Bruprosjektering",
        }
    ]


@pytest.mark.parametrize("reader", ["materialize", "structure", "profiles"])
def test_the_flow_form_is_returned_unchanged(reader: str, tmp_path: Path) -> None:
    flat = _flat_readings(FLOW, tmp_path)[reader]
    assert flat["sources"] == "[{ resource: a.pdf, title: A }]"
    assert flat["generated"] == "{ by: process:okf-ingest, at: 2026-08-31T00:00:00Z }"


@pytest.mark.parametrize("reader", ["materialize", "structure", "profiles"])
def test_an_absent_key_stays_absent(reader: str, tmp_path: Path) -> None:
    assert "sources" not in _flat_readings(NO_SOURCES, tmp_path)[reader]
    assert _read_sources(NO_SOURCES, tmp_path) == ((), False)


@pytest.mark.parametrize("reader", ["materialize", "structure", "profiles"])
def test_an_indented_line_before_a_dash_invents_no_entry(reader: str, tmp_path: Path) -> None:
    """`read_sources` refuses this shape; the flat readers reach the same
    conclusion. The flat grammar has no third state, so "nothing decoded" is
    the empty value it already returned -- unchanged, and stated rather than
    silently improved."""
    assert _read_sources(INDENTED_BEFORE_DASH, tmp_path) == ((), True)
    assert _flat_readings(INDENTED_BEFORE_DASH, tmp_path)[reader]["sources"] == ""


@pytest.mark.parametrize("reader", ["materialize", "structure", "profiles"])
def test_quoted_leaves_follow_the_k3_22_rule(reader: str, tmp_path: Path) -> None:
    """A `"`-wrapped leaf is decoded, a `'`-wrapped one stands -- `read_sources`'
    rule, which is where the entries come from. PyYAML decodes both and the
    guard decodes neither; the divergence is named rather than averaged."""
    value = _flat_readings(QUOTED_LEAVES, tmp_path)[reader]["sources"]
    entries, _present = _read_sources(QUOTED_LEAVES, tmp_path)
    assert _parse_flow_mappings(value) == [dict(entry) for entry in entries]
    assert _parse_flow_mappings(value) == [{"resource": "a, b.pdf", "title": "'N100'"}]


# --- the shipped fixtures, all of them, not a sample ----------------------


def _fixture_frontmatters() -> list[Path]:
    found = []
    for path in sorted(FIXTURES.rglob("*.md")):
        first = path.read_text(encoding="utf-8").splitlines()[:1]
        if first and first[0].strip() == "---":
            found.append(path)
    return found


def test_every_fixture_frontmatter_keeps_every_value_a_reference_reader_finds() -> None:
    """The defect stated as an invariant over the shipped fixtures: no flat
    reader may return an EMPTY value for a scalar or a `sources` block that
    PyYAML reads as a value, and all three copies must agree key for key.

    The denominator is pinned so a refactor that stops finding the fixtures
    cannot turn this green over an empty set."""
    paths = _fixture_frontmatters()
    assert len(paths) == 12
    carrying_sources_list = 0
    for path in paths:
        text = path.read_text(encoding="utf-8")
        flat = parse_frontmatter(path)
        assert flat == _structure_split(text)[0]
        assert flat == _profiles_split(text)[0]
        block = text.split("---\n")[1]
        reference = yaml.safe_load(block)
        assert isinstance(reference, dict)
        for key, value in reference.items():
            if value in (None, "", [], {}):
                continue
            if isinstance(value, (list, dict)) and key not in STRUCTURED_BLOCK_KEYS:
                continue
            assert flat.get(key, "") != "", f"{path.name}: {key} lost"
        if isinstance(reference.get("sources"), list):
            carrying_sources_list += 1
            assert _parse_flow_mappings(flat["sources"]) == reference["sources"]
    # Two of the twelve: one block form and one flow form, both read by
    # PyYAML as a list and both required to decode to the same entries here.
    assert carrying_sources_list == 2


def test_a_block_key_outside_the_named_set_is_still_empty() -> None:
    """The limit of this fix, pinned rather than left to be discovered.

    The named set is one key wide on purpose: `sources` is the key
    `read_sources` already knows how to read, and widening the flat grammar to
    every block collection would change what this library reports for keys no
    measurement covers. A fixture in this tree carries a block `verified:` for
    exactly this reason, and it still reads as an empty value. Whoever widens
    the set will see this test, which is the point of pinning it."""
    assert STRUCTURED_BLOCK_KEYS == frozenset({"sources"})
    path = FIXTURES / "consume-bundle" / "dyp" / "nivaa" / "blokkform-verifisert.md"
    reference = yaml.safe_load(path.read_text(encoding="utf-8").split("---\n")[1])
    assert isinstance(reference["verified"], list)
    assert parse_frontmatter(path)["verified"] == ""