llm-ingestion-okf/tests/test_run_frontmatter.py

"""A run can state keys for every concept it writes (K3-19 b).

MEASURED OUTSIDE THIS REPOSITORY: two bundles of the same kind of source, one
built by `okf build` and one by a consumer's own script. The second carries
`description`, an edition key and a `sources` entry whose address is the
publisher's URL -- eight keys of its own -- and passes `okf check` with 0
findings. The first could carry none of them without a line of Python.

SPEC SS 4.1 "Extensions", verbatim: "Producers MAY include any additional
keys. Consumers SHOULD preserve unknown keys when round-tripping and MUST NOT
reject documents with unrecognized fields." SS 11 lists "Unknown additional
frontmatter keys" among what a consumer MUST NOT reject a bundle for.

THE PRECEDENCE: a value stated for the run beats what the document declares,
which beats the file name. `sources` and `description` are the two keys with a
declared layer below the flag, so they are the two a run may REPLACE. Every
other key this door writes is refused: it is either measured from the bytes
(`source_sha256`, the offsets, the locators), owned by another flag (`type`,
`ingested_at`, `bundle_id`), the ownership stamp a later run reads back
(`generated`), or a derived facet whose `derived` marker would go on naming a
value the flag had replaced.

THE FORM IS NOT A STYLE CHOICE. The value is written verbatim on ONE line,
because this package's own readers are line-oriented and measured blind to a
block-form `sources`. The flag splits on the FIRST `=` and only there: the
value a publisher's address needs carries `=` itself.
"""

from __future__ import annotations

from pathlib import Path

import pytest

from llm_ingestion_okf import cli
from llm_ingestion_okf.consume import read_sources
from llm_ingestion_okf.errors import IngestError
from llm_ingestion_okf.materialize import parse_frontmatter

FIXTURES = Path(__file__).parent / "fixtures"
IDENTITY = FIXTURES / "sts-identity.xml"
OPAQUE = "0f1e2d3c-4b5a-6978-8796-a5b4c3d2e1f0.xml"

ADDRESS = "https://normer.example.test/api/sts/900?languageCode=nb"
STATED_SOURCES = f"[{{ resource: {ADDRESS}, title: R900:2024 }}]"
MARKDOWN = b"# Innledning\n\nTekst her.\n\n# Omfang\n\nMer tekst her.\n"


def _inbox(tmp_path: Path) -> Path:
    inbox = tmp_path / "docs"
    inbox.mkdir()
    (inbox / OPAQUE).write_bytes(IDENTITY.read_bytes())
    (inbox / "notat.md").write_bytes(MARKDOWN)
    return inbox


def _main(inbox: Path, bundle: Path, *extra: str) -> int:
    return cli.main(
        [
            "build",
            str(inbox),
            "--bundle",
            str(bundle),
            "--bundle-id",
            "run",
            "--okf-version",
            "0.2",
            *extra,
        ]
    )


def _concept_files(bundle: Path) -> list[Path]:
    return [
        path for path in sorted(bundle.rglob("*.md")) if path.name not in ("index.md", "log.md")
    ]


def _frontmatter_lines(path: Path) -> list[str]:
    lines = path.read_text(encoding="utf-8").splitlines()
    return lines[1 : lines.index("---", 1)]


# --- the flag's grammar ----------------------------------------------------


def test_the_flag_splits_on_the_first_equals_sign_and_only_there() -> None:
    assert cli.frontmatter_from_flags(
        [f"sources={STATED_SOURCES}", "utgave=R900:2024", "merknad=a=b"]
    ) == {"sources": STATED_SOURCES, "utgave": "R900:2024", "merknad": "a=b"}


def test_a_flag_without_an_equals_sign_is_refused() -> None:
    with pytest.raises(IngestError) as caught:
        cli.frontmatter_from_flags(["utgave"])
    assert caught.value.code == "run_frontmatter_invalid"


def test_the_same_key_twice_is_refused_rather_than_resolved() -> None:
    """Which of two values wins would be a guess about what the caller meant."""
    with pytest.raises(IngestError) as caught:
        cli.frontmatter_from_flags(["utgave=A", "utgave=B"])
    assert caught.value.code == "run_frontmatter_invalid"


# --- what lands in the bundle ----------------------------------------------


def test_every_concept_carries_the_stated_keys_once(tmp_path: Path) -> None:
    bundle = tmp_path / "bundle"
    assert (
        _main(
            _inbox(tmp_path),
            bundle,
            "--frontmatter",
            "utgave=R900:2024",
            "--frontmatter",
            f"sources={STATED_SOURCES}",
        )
        == 0
    )
    concepts = _concept_files(bundle)
    assert len({path.relative_to(bundle).parts[0] for path in concepts}) == 2, (
        "both documents -- the STS one and the markdown one -- must produce concepts"
    )
    for path in concepts:
        lines = _frontmatter_lines(path)
        assert lines.count("utgave: R900:2024") == 1, path
        # ONE `sources` line, and it is the stated one: the flag REPLACES what
        # the door would derive -- the document's own title on the STS file,
        # the file name on the markdown one -- and never adds a second.
        assert [line for line in lines if line.startswith("sources:")] == [
            f"sources: {STATED_SOURCES}"
        ], path


def test_the_stated_address_reads_back_through_this_packages_own_readers(
    tmp_path: Path,
) -> None:
    bundle = tmp_path / "bundle"
    assert _main(_inbox(tmp_path), bundle, "--frontmatter", f"sources={STATED_SOURCES}") == 0
    for path in _concept_files(bundle):
        assert parse_frontmatter(path)["sources"] == STATED_SOURCES
        assert read_sources(_frontmatter_lines(path)) == (
            ({"resource": ADDRESS, "title": "R900:2024"},),
            True,
        )


def test_the_flag_adds_its_line_and_moves_nothing_else(tmp_path: Path) -> None:
    """Without the flag the bytes are today's; with it, exactly one line more."""
    inbox = _inbox(tmp_path)
    plain, stamped = tmp_path / "plain", tmp_path / "stamped"
    assert _main(inbox, plain) == 0
    assert _main(inbox, stamped, "--frontmatter", "utgave=R900:2024") == 0

    before = {
        p.relative_to(plain).as_posix(): p.read_text(encoding="utf-8")
        for p in sorted(plain.rglob("*"))
        if p.is_file()
    }
    after = {
        p.relative_to(stamped).as_posix(): p.read_text(encoding="utf-8")
        for p in sorted(stamped.rglob("*"))
        if p.is_file()
    }
    assert before.keys() == after.keys()
    moved = [name for name in before if before[name] != after[name]]
    assert moved, "the flag must reach at least one concept for this control to mean anything"
    for name in moved:
        assert after[name].replace("utgave: R900:2024\n", "", 1) == before[name], name


def test_the_unsegmented_route_carries_them_too(tmp_path: Path) -> None:
    bundle = tmp_path / "bundle"
    assert (
        _main(_inbox(tmp_path), bundle, "--segments", "off", "--frontmatter", "utgave=R900:2024")
        == 0
    )
    concepts = _concept_files(bundle)
    assert concepts
    assert all(parse_frontmatter(path)["utgave"] == "R900:2024" for path in concepts)


# --- what is refused, before anything is written -----------------------------


@pytest.mark.parametrize(
    "key",
    [
        "type",
        "title",
        "source_file",
        "source_sha256",
        "ingested_at",
        "generated",
        "bundle_id",
        "segment_id",
        "source_offset",
        "source_lines",
        "source_pages",
        "adjudication",
        "parent",
        "number",
        "derived",
    ],
)
def test_a_key_the_door_writes_itself_is_refused(
    tmp_path: Path, key: str, capsys: pytest.CaptureFixture[str]
) -> None:
    bundle = tmp_path / "bundle"
    assert _main(_inbox(tmp_path), bundle, "--frontmatter", f"{key}=X") == 2
    assert key in capsys.readouterr().err
    assert not bundle.exists()


@pytest.mark.parametrize(
    "pair",
    ["ut gave=X", "-utgave=X", "ut:gave=X", "=X", "utgave=", "utgave= X", "utgave=X "],
    ids=["space-in-key", "dash-first", "colon-in-key", "no-key", "empty", "lead", "trail"],
)
def test_a_pair_that_would_not_read_back_verbatim_is_refused(tmp_path: Path, pair: str) -> None:
    bundle = tmp_path / "bundle"
    # The attached `--frontmatter=PAIR` form: argparse reads a detached
    # argument that starts with `-` as an option of its own.
    assert _main(_inbox(tmp_path), bundle, f"--frontmatter={pair}") == 2
    assert not bundle.exists()


def test_a_value_spanning_lines_is_refused_through_the_api(tmp_path: Path) -> None:
    with pytest.raises(IngestError) as caught:
        cli.build(
            _inbox(tmp_path),
            tmp_path / "bundle",
            bundle_id="run",
            okf_version="0.2",
            frontmatter={"utgave": "R900\nsources: [{ resource: x }]"},
        )
    assert caught.value.code == "run_frontmatter_invalid"
    assert not (tmp_path / "bundle").exists()