portfolio-optimiser-claude/tests/test_ingest_spec_loadbearing.py

"""Spec-integrity seam for the ingest spec (ingest-spec §11).

The D7 analog of MAF's I1 framework-guard: this repo consumes ``shared/ingest-spec.md``
UNCHANGED from commons, and this test keeps the contract honest — it goes RED when the
spec goes missing, names a concrete agent toolkit (the framework-neutrality rule), or
stops documenting a contract field. It is the load-bearing guard the ingest layer relies
on to keep being implementable "from this spec alone".

Form mirrored from the sibling ``test_method_spec_loadbearing.py``: every predicate takes
the spec TEXT as an argument, so the detach-proofs are tests in the suite rather than a
one-off spot-check that dies with the session. Red-proofs run against a mutated COPY of
the spec in ``tmp_path`` — never against ``shared/`` itself.
"""

from __future__ import annotations

from pathlib import Path

import pytest

SPEC = Path(__file__).resolve().parents[1] / "shared" / "ingest-spec.md"

# Concrete agent toolkits / vendor stacks the framework-neutral spec MUST NOT name.
_FORBIDDEN_TOOLKITS = (
    "claude",
    "anthropic",
    "openai",
    "gpt",
    "gemini",
    "llama",
    "langchain",
    "autogen",
    "crewai",
    "semantic kernel",
    "microsoft agent framework",
    "agent sdk",
    "bedrock",
    "vertex",
    "foundry",
    "maf",
)

# The §12 cross-check table is the ANCHOR, and the spec appoints it itself: "Every field
# of the machine-readable contracts, mapped to its normative section (completeness is
# enforced by the spec-integrity test)" — this test is that enforcer. Asserting over the
# whole spec text instead would be green-but-dead: prose saturates the field names (§1's
# honesty rule alone carries `generated: true` twice), so no amendment dropping a row
# could ever turn it red.
_CROSS_CHECK_HEADING = "## 12. Cross-check table"

# Every field of the machine-readable contracts the D7 implementation depends on — the
# spec's §12 cross-check table must keep documenting each (spec-integrity).
_CONTRACT_FIELDS = (
    "manifest_version",
    "source",
    "bundle_summary",
    "extractions",
    "source_system",
    "source_query",
    "ingested_at",
    "ingest_manifest",
    "generated",
    "okf_type",
    "max_rows",
    "root",
    "connection_ref",  # the sql source reference the D7 sql connector (I5) depends on
)


def _cross_check_table(text: str) -> str:
    """The §12 section body — heading to end-of-spec or the next section, whichever first."""
    start = text.index(_CROSS_CHECK_HEADING)  # RED (ValueError) if §12 is renamed or dropped
    end = text.find("\n## ", start + len(_CROSS_CHECK_HEADING))
    return text[start:] if end == -1 else text[start:end]


def _named_toolkits(text: str) -> list[str]:
    low = text.lower()
    return [tok for tok in _FORBIDDEN_TOOLKITS if tok in low]


def _undocumented_fields(text: str) -> list[str]:
    # The row's FIRST column is the documented-field claim; a field named only in another
    # row's "Contract" prose does not count.
    table = _cross_check_table(text)
    return [field for field in _CONTRACT_FIELDS if f"| `{field}` |" not in table]


def _slice_defects(table: str) -> list[str]:
    """Defects in the anchor itself — a slice that widened into the surrounding spec."""
    defects = []
    if not table.startswith(_CROSS_CHECK_HEADING):
        defects.append("does not start at the §12 heading")
    if "Honesty rule" in table:
        defects.append("leaked §1 prose — the anchor is not a slice")
    if "\n## " in table:
        defects.append("leaked a following section")
    return defects


def _row_removed(text: str, field: str) -> str:
    table = _cross_check_table(text)
    kept = [ln for ln in table.splitlines(keepends=True) if not ln.startswith(f"| `{field}` |")]
    return text.replace(table, "".join(kept))


def _row_renamed(text: str, field: str) -> str:
    table = _cross_check_table(text)
    return text.replace(table, table.replace(f"| `{field}` |", f"| `{field}_renamed` |"))


# --- The guard itself (against the real spec) ---------------------------------------


def test_spec_is_present() -> None:
    # RED if the spec goes missing (the layer stops being implementable from spec alone).
    assert SPEC.is_file(), "ingest-spec.md missing — subtree pull the commons contract"


def test_spec_names_no_agent_toolkit() -> None:
    present = _named_toolkits(SPEC.read_text(encoding="utf-8"))
    assert not present, f"framework-neutral spec names a concrete toolkit: {present}"


def test_cross_check_slice_is_a_slice_and_not_the_whole_spec() -> None:
    # Guards the anchor itself: a slice that degenerated into the full text would make
    # every row assertion below green-but-dead again, silently. RED if it widens.
    defects = _slice_defects(_cross_check_table(SPEC.read_text(encoding="utf-8")))
    assert not defects, f"the §12 anchor degenerated: {defects}"


@pytest.mark.parametrize("field", _CONTRACT_FIELDS)
def test_spec_documents_contract_field(field: str) -> None:
    # RED when an amendment drops or renames a §12 row.
    undocumented = _undocumented_fields(SPEC.read_text(encoding="utf-8"))
    assert field not in undocumented, (
        f"contract field {field!r} is no longer a row in the §12 cross-check table"
    )


# --- Red-proofs: the guard MUST fail on a detached spec (mutated copy, never shared/) --


def test_guard_red_when_spec_missing(tmp_path: Path) -> None:
    assert not (tmp_path / "ingest-spec.md").is_file()


@pytest.mark.parametrize("toolkit", _FORBIDDEN_TOOLKITS)
def test_guard_red_when_toolkit_injected(tmp_path: Path, toolkit: str) -> None:
    mutated = SPEC.read_text(encoding="utf-8") + f"\n\nBuilt on {toolkit}.\n"
    copy = tmp_path / "ingest-spec.md"
    copy.write_text(mutated, encoding="utf-8")
    assert toolkit in _named_toolkits(copy.read_text(encoding="utf-8"))


@pytest.mark.parametrize("field", _CONTRACT_FIELDS)
def test_guard_red_when_row_removed_from_cross_check_table(tmp_path: Path, field: str) -> None:
    # M1: the row is gone from §12 — even though prose elsewhere may still name the field.
    copy = tmp_path / "ingest-spec.md"
    copy.write_text(_row_removed(SPEC.read_text(encoding="utf-8"), field), encoding="utf-8")
    assert field in _undocumented_fields(copy.read_text(encoding="utf-8"))


@pytest.mark.parametrize("field", _CONTRACT_FIELDS)
def test_guard_red_when_row_renamed_in_cross_check_table(tmp_path: Path, field: str) -> None:
    # M2: a detach-proof is not a value-proof — the row still exists, under another name.
    copy = tmp_path / "ingest-spec.md"
    copy.write_text(_row_renamed(SPEC.read_text(encoding="utf-8"), field), encoding="utf-8")
    assert field in _undocumented_fields(copy.read_text(encoding="utf-8"))


def test_guard_red_when_section12_heading_renamed(tmp_path: Path) -> None:
    # M3: fail-closed — no §12 heading means no anchor, and the guard must raise, not
    # silently fall back to a wider (green-but-dead) slice.
    mutated = SPEC.read_text(encoding="utf-8").replace(
        _CROSS_CHECK_HEADING, "## 12. Field reference"
    )
    copy = tmp_path / "ingest-spec.md"
    copy.write_text(mutated, encoding="utf-8")
    with pytest.raises(ValueError):
        _cross_check_table(copy.read_text(encoding="utf-8"))


def test_slice_guard_red_when_anchor_degenerates_to_whole_spec() -> None:
    # M4: the anchor can degenerate. If _cross_check_table ever returned the full text,
    # the row assertions would go green-but-dead again — this proves the slice guard is
    # what catches that, and that it is not itself green by accident.
    defects = _slice_defects(SPEC.read_text(encoding="utf-8"))
    assert defects, "the slice guard accepts the whole spec as the §12 table — it is dead"