llm-ingestion-okf/tests/test_okf_v0_2_profile.py

"""D2 — the `OKF_V0_2` profile, profile-aware ownership, `sources` derivation.

Three things, in the order the risk runs:

1. **The profile constant.** Names the §5 families, requires only `type` (§4:
   "the only always-required key"), and closes nothing — §14 forbids a consumer
   to reject on unknown types or unknown keys, so an allowlist here would put
   the profile in violation of the spec it is named after.
2. **Ownership.** `_is_ingest_owned` is the pre-mutation collision gate, so it
   is the one piece of this work whose defects are expensive and quiet. The v0.2
   stamp is a flow mapping carrying `ingested_at`, so equality is impossible by
   construction and the predicate is a prefix test. The operator decided the
   direction: `OKF_V0_2` accepts BOTH stamps, so a bundle written under `DEFAULT`
   re-runs IN PLACE; `DEFAULT` does not accept the v0.2 form, so the recognition
   is one-way and the fail-safe from V-A3 is preserved.
3. **`sources`.** Derived from the manifest's source as an inline flow sequence
   (requirement 1's measured form), with the locator taken verbatim per source
   type: `root`, `connection_ref`, `base_url`. `credential_ref` is never a
   locator and never reaches frontmatter.

`generated.by` is `process:llm-ingestion-okf` — option (d) from the plan's V1,
chosen by the operator 2026-07-27 over the `<producer>/<version>` form A-E3 was
written against. The value carries no version, which is what keeps a byte-
compared fixture stable across releases. **A-E3 is now stale in the pilot
specification and owes po-claude a correction.**
"""

from __future__ import annotations

import re
from pathlib import Path

import pytest

from llm_ingestion_okf.errors import MaterializationError
from llm_ingestion_okf.manifest import (
    Extraction,
    FileSource,
    HttpSource,
    Manifest,
    Source,
    SqlSource,
)
from llm_ingestion_okf.materialize import _is_ingest_owned, _render_concept_file
from llm_ingestion_okf.profiles import DEFAULT, OKF_LATEST, OKF_V0_2, STRICT_V1, BundleProfile

INGESTED_AT = "2026-07-25T12:00:00Z"
STAMP = "manifest@0123456789abcdef"
V0_2_STAMP = f"{{ by: process:llm-ingestion-okf, at: {INGESTED_AT} }}"


def _manifest(source: Source) -> Manifest:
    return Manifest(
        manifest_version=1,
        source=source,
        bundle_summary="A test bundle.",
        extractions=(
            Extraction(
                id="orders", title="Orders", query="orders.csv", okf_type="dataset", max_rows=10
            ),
        ),
    )


def _concept(source: Source, *, profile: BundleProfile = OKF_V0_2) -> str:
    manifest = _manifest(source)
    return _render_concept_file(
        manifest,
        manifest.extractions[0],
        "Body.\n",
        ingested_at=INGESTED_AT,
        stamp=STAMP,
        profile=profile,
    )


def _stamped_concept(generated: str) -> bytes:
    return (
        "---\ntype: dataset\ntitle: Orders\n"
        f"ingest_manifest: {STAMP}\n"
        f"generated: {generated}\n---\n\nBody.\n"
    ).encode()


# --- the profile constant -------------------------------------------------


def test_the_v0_2_profile_names_the_five_families_it_must_express() -> None:
    """C-E3's list, as emission order rather than as prose. Naming a key is what
    lets a caller emit it in canonical order; judging a structured value is D1's
    reader and deliberately absent here.
    """
    order = OKF_V0_2.frontmatter.order

    for key in ("generated", "sources", "verified", "status", "stale_after"):
        assert key in order
    assert "timestamp" not in order  # §13.1: the fallback is for v0.1 documents
    assert order.index("generated") < order.index("sources")


def test_the_v0_2_profile_requires_only_type() -> None:
    """§4/§11: `type` is the only always-required key, and a concept carrying
    just `type` is fully conformant."""
    assert OKF_V0_2.frontmatter.required == frozenset({"type"})


def test_the_v0_2_profile_closes_neither_the_type_set_nor_the_key_namespace() -> None:
    """§14's MUST NOTs, expressed as configuration: a consumer must not reject
    on an unknown `type` value or on unknown additional keys. A profile that
    closed either would be non-conformant against the version it is named for.
    """
    assert OKF_V0_2.types.allowed is None
    assert OKF_V0_2.frontmatter.allowed is None
    assert OKF_V0_2.frontmatter.key_pattern is None


def test_the_v0_2_profile_keeps_the_source_query_collapse_and_the_default_paths() -> None:
    """The parts that are commons' §5 rather than upstream's: the doors write
    the same filenames whatever the OKF version, and a multi-line SELECT still
    renders on one line."""
    assert OKF_V0_2.frontmatter.collapsed_keys == frozenset({"source_query"})
    assert OKF_V0_2.paths == DEFAULT.paths


def test_okf_latest_still_means_v0_1_until_ga() -> None:
    """`OKF_LATEST` is "the latest version supported as STABLE", not the latest
    present in the tree. Flipping it is the GA event — a single auditable
    action rather than a side effect of this merge — so during the pilot a
    consumer bound to the alias keeps v0.1 semantics.
    """
    assert OKF_LATEST is DEFAULT


# --- ownership ------------------------------------------------------------


def test_the_default_profile_still_stamps_the_v0_1_literal() -> None:
    assert DEFAULT.ownership.stamp(INGESTED_AT) == "true"


def test_the_v0_2_stamp_is_a_flow_mapping_whose_at_is_the_argument_exactly() -> None:
    """A-E3, corrected: the actor is the §7 `process:<id>` form and carries no
    version. `at` binds to the already-validated `ingested_at` argument, so no
    wall-clock is introduced and two runs with the same argument agree.
    """
    assert OKF_V0_2.ownership.stamp(INGESTED_AT) == V0_2_STAMP
    assert "llm-ingestion-okf/" not in OKF_V0_2.ownership.stamp(INGESTED_AT)


def test_the_v0_2_predicate_accepts_the_stamp_its_own_profile_writes(tmp_path: Path) -> None:
    """A-E5's second run: a v0.2 bundle re-runs into the directory it wrote."""
    path = tmp_path / "ingest-orders.md"
    path.write_bytes(_stamped_concept(V0_2_STAMP))

    assert _is_ingest_owned(path, "manifest", profile=OKF_V0_2) is True


def test_the_v0_2_predicate_also_accepts_the_v0_1_stamp(tmp_path: Path) -> None:
    """The operator's decision (2026-07-27): a `DEFAULT`-written bundle is
    re-runnable IN PLACE under `OKF_V0_2`. Without this the profile switch would
    fire `collision_unstamped` on the library's own previous output, and the
    black-box promise — an upstream release costs a consumer a re-run, nothing
    more — would not hold across the profile change.
    """
    path = tmp_path / "ingest-orders.md"
    path.write_bytes(_stamped_concept("true"))

    assert _is_ingest_owned(path, "manifest", profile=OKF_V0_2) is True


def test_the_default_predicate_still_refuses_the_v0_2_stamp(tmp_path: Path) -> None:
    """Recognition is one-way, which is what preserves V-A3. `DEFAULT` meeting a
    v0.2 file refuses the run rather than replacing the file."""
    path = tmp_path / "ingest-orders.md"
    path.write_bytes(_stamped_concept(V0_2_STAMP))

    assert _is_ingest_owned(path, "manifest") is False
    assert _is_ingest_owned(path, "manifest", profile=DEFAULT) is False


def test_a_foreign_v0_2_actor_is_never_ingest_owned(tmp_path: Path) -> None:
    """U3 as a test: `generated` is attribution, and upstream writes it for
    HAND-AUTHORED files. Presence proves nothing — the actor has to be ours, or
    the collision gate would claim a human's file that merely carries the key.
    """
    path = tmp_path / "ingest-orders.md"
    path.write_bytes(_stamped_concept(f"{{ by: human:jsmith@acme, at: {INGESTED_AT} }}"))

    assert _is_ingest_owned(path, "manifest", profile=OKF_V0_2) is False


def test_the_v0_2_stamp_still_has_to_name_this_manifest(tmp_path: Path) -> None:
    """§10.2 per-manifest ownership is unchanged by the profile: a sibling
    manifest sharing the bundle keeps its own files."""
    path = tmp_path / "ingest-orders.md"
    path.write_bytes(_stamped_concept(V0_2_STAMP))

    assert _is_ingest_owned(path, "other-manifest", profile=OKF_V0_2) is False


# --- `sources` derivation -------------------------------------------------


SOURCE_CASES = [
    pytest.param(FileSource(id="golden-catalogue", root="fixture"), "fixture", id="file"),
    pytest.param(
        SqlSource(id="golden-db", connection_ref="OKF_GOLDEN_SQL_DB"),
        "OKF_GOLDEN_SQL_DB",
        id="sql",
    ),
    pytest.param(
        HttpSource(id="golden-api", base_url="https://golden.example.test"),
        "https://golden.example.test",
        id="http",
    ),
]


@pytest.mark.parametrize("source,resource", SOURCE_CASES)
def test_sources_is_an_inline_flow_sequence_carrying_the_locator(
    source: Source, resource: str
) -> None:
    """A-E4. One entry, `id` from the manifest source and `resource` its locator
    verbatim — the filesystem root, the env-var NAME holding the DSN, or the base
    URL. Upstream's `resource` is a bundle-internal path; ours is not, and
    inventing the `author`/`last_modified` upstream also carries would be writing
    fields with no reader.
    """
    line = f"sources: [{{ id: {source.id}, resource: {resource} }}]"

    assert line in _concept(source).splitlines()


def test_sources_follows_generated_in_the_emitted_block() -> None:
    """The order the schema declares is the order that reaches disk."""
    lines = _concept(FileSource(id="golden-catalogue", root="fixture")).splitlines()

    assert [line.split(":")[0] for line in lines[1:9]] == [
        "type",
        "title",
        "source_system",
        "source_query",
        "ingested_at",
        "ingest_manifest",
        "generated",
        "sources",
    ]


def test_the_default_profile_emits_neither_sources_nor_a_v0_2_generated() -> None:
    """V-A6 at the unit level: the seven-key v0.1 block, byte for byte. The
    golden suite proves the same thing end to end; this one localizes a failure
    to the renderer rather than to a fixture diff.
    """
    rendered = _concept(FileSource(id="golden-catalogue", root="fixture"), profile=DEFAULT)

    assert rendered == (
        "---\n"
        "type: dataset\n"
        "title: Orders\n"
        "source_system: golden-catalogue\n"
        "source_query: orders.csv\n"
        f"ingested_at: {INGESTED_AT}\n"
        f"ingest_manifest: {STAMP}\n"
        "generated: true\n"
        "---\n"
        "\n"
        "Body.\n"
    )


def test_the_http_credential_reference_never_reaches_the_frontmatter() -> None:
    """A credential reference is not a locator. The convention forbids the
    credential itself in frontmatter; the env-var NAME holding it has no reader
    in a bundle either, so it is not written.
    """
    source = HttpSource(
        id="golden-api",
        base_url="https://golden.example.test",
        credential_ref="OKF_GOLDEN_API_TOKEN",
    )

    assert "OKF_GOLDEN_API_TOKEN" not in _concept(source)


@pytest.mark.parametrize(
    "root",
    ["data, backup", "data{1}", "data[1]", "key: value"],
    ids=["comma", "braces", "brackets", "colon-space"],
)
def test_a_locator_that_would_restructure_the_flow_mapping_is_refused(root: str) -> None:
    """Fail-fast rather than emit a silently WRONG provenance record.

    These characters terminate or restructure a YAML flow mapping, and the
    failure is quiet rather than loud. Measured with PyYAML 2026-07-27 rather
    than argued: `[{ id: a, resource: data, backup }]` raises nothing and yields
    `[{'id': 'a', 'resource': 'data', 'backup': None}]` — a clean parse into a
    record no one wrote. Refusing is the same posture as the filename-length
    gate: validation, not repair.

    The same run confirms what IS emitted parses as intended — `generated` to a
    two-key mapping, `sources` to a one-element list of one — including a base
    URL, whose colons stay inside a plain scalar in flow context.
    """
    with pytest.raises(MaterializationError) as excinfo:
        _concept(FileSource(id="golden-catalogue", root=root))

    assert excinfo.value.code == "source_reference_unquotable"


def test_the_default_profile_is_unaffected_by_an_unquotable_locator() -> None:
    """The refusal is a property of the v0.2 emission, not a new gate on v0.1:
    `DEFAULT` never writes the locator, so it has nothing to refuse. A shared
    refusal would have broken additivity — a manifest that ran yesterday must
    still run."""
    rendered = _concept(FileSource(id="golden-catalogue", root="data, backup"), profile=DEFAULT)

    assert "data, backup" not in rendered


# --- V-A5 extended to the new profile -------------------------------------

_VERSION_LITERAL_RE = re.compile(r"^\d+(\.\d+)+$")


@pytest.mark.parametrize(
    "profile", [DEFAULT, STRICT_V1, OKF_V0_2], ids=["DEFAULT", "STRICT_V1", "OKF_V0_2"]
)
def test_no_profile_hard_codes_an_upstream_version(profile: BundleProfile) -> None:
    """V-A5, restated over the profile the name of which is the one place a
    version literal would look natural. `OKF_V0_2` is a NAME; the value belongs
    to catalog (E1) and is declared at D5, in the fixture, once.

    Kept alongside the characterization suite's copy rather than replacing it:
    that one pins the two shipped profiles at the point they were measured.
    """
    from test_okf_v0_2_characterization import _strings_in

    assert sorted({s for s in _strings_in(profile) if _VERSION_LITERAL_RE.match(s)}) == []