llm-ingestion-pipeline-security/tests/test_catalog_register.py

"""Every catalog the guard cites carries its edition, a check date and a source link — row 1 of the 2026 done-criterion.

Row 1 (`docs/PLAN.md` § T.2) is N of N register entries on the newest edition, each
with the date it was last checked and a link to where it was checked. The register
is `docs/threat-2026/catalogs.json`; its denominator was frozen at 15 entries in
session 1 and may grow, never shrink, without the operator's yes.

This file fails when an entry lacks a date or a link, or cites an edition other
than the one recorded as latest. It does not fail on age: that is the release
rule in `tests/test_release_rule.py` (direction of 2026-11-01), which judges the
check dates against the release date, so the suite does not turn red merely
because a day passed between releases.

The README may only name an OWASP Agentic or MITRE ATLAS entry the register
records as read at the publisher, and must name the ATLAS release the register
records as latest — that is how "what is not verified is left out" stays true
after the session that wrote it. No network: the publisher reads are recorded in
the register's ``evidence`` fields, not repeated here.
"""
from __future__ import annotations

import datetime
import json
import re
from pathlib import Path
from urllib.parse import urlsplit

ROOT = Path(__file__).resolve().parent.parent
REGISTER = ROOT / "docs" / "threat-2026" / "catalogs.json"
README = ROOT / "README.md"

# The denominator frozen 2026-10-28 (session 1). Entries may be added; none of
# these may disappear without the operator's yes.
FROZEN_IDS = frozenset({
    "owasp-llm-top10", "owasp-agentic-top10", "mitre-atlas", "owasp-rag-cheat-sheet",
    "unicode", "cve-2025-32711", "arxiv-2402.07867", "arxiv-2504.11168",
    "arxiv-2509.14285", "arxiv-2505.03574", "arxiv-2503.18813", "arxiv-2604.00387",
    "willison-dual-llm", "rfc-3986", "llm-security-commons",
})

LINK_SCHEMES = frozenset({"https", "ssh"})


def _register() -> dict:
    return json.loads(REGISTER.read_text(encoding="utf-8"))


def _entries() -> list[dict]:
    return _register()["entries"]


def _problems(entry: dict) -> list[str]:
    """Why an entry does not count toward row 1; empty when it does."""
    problems = []
    for field in ("id", "name", "publisher", "latest", "cited", "checked", "source"):
        if not isinstance(entry.get(field), str) or not entry[field].strip():
            problems.append(f"missing {field}")
    checked = entry.get("checked")
    if isinstance(checked, str) and checked.strip():
        try:
            datetime.date.fromisoformat(checked)
        except ValueError:
            problems.append(f"checked is not an ISO date: {checked!r}")
    source = entry.get("source")
    if isinstance(source, str) and source.strip():
        parts = urlsplit(source)
        if parts.scheme not in LINK_SCHEMES or not parts.netloc:
            problems.append(f"source is not a link: {source!r}")
    if entry.get("cited") != entry.get("latest"):
        problems.append(f"cites {entry.get('cited')!r}, latest is {entry.get('latest')!r}")
    return problems


def test_the_check_goes_red_on_a_missing_date_a_missing_link_or_an_old_edition():
    good = {"id": "x", "name": "X", "publisher": "P", "latest": "2", "cited": "2",
            "checked": "2026-10-31", "source": "https://example.test/x"}
    assert _problems(good) == []
    assert "missing checked" in _problems({**good, "checked": ""})
    assert any("ISO date" in p for p in _problems({**good, "checked": "31.10.2026"}))
    assert "missing source" in _problems({k: v for k, v in good.items() if k != "source"})
    assert any("not a link" in p for p in _problems({**good, "source": "arXiv 2402.07867"}))
    assert any("latest is" in p for p in _problems({**good, "cited": "1"}))


def test_every_entry_has_a_check_date_a_source_link_and_the_latest_edition():
    failing = {e.get("id", "?"): _problems(e) for e in _entries()}
    failing = {k: v for k, v in failing.items() if v}
    assert not failing, "register entries that do not count toward row 1:\n" + "\n".join(
        f"  {k}: {'; '.join(v)}" for k, v in failing.items())


def test_the_frozen_denominator_never_shrinks():
    ids = [e["id"] for e in _entries()]
    assert len(ids) == len(set(ids)), "duplicate register ids"
    missing = FROZEN_IDS - set(ids)
    assert not missing, f"entries removed without the operator's yes: {sorted(missing)}"


def test_unverified_candidates_are_kept_apart_from_the_count():
    # A source the repo cites but that could not be read at its publisher is
    # recorded as a candidate, never as an entry with a borrowed date.
    entry_ids = {e["id"] for e in _entries()}
    for candidate in _register().get("candidates", []):
        assert candidate["id"] not in entry_ids
        assert candidate.get("why_not_counted", "").strip()


def _entry(entry_id: str) -> dict:
    (entry,) = [e for e in _entries() if e["id"] == entry_id]
    return entry


def test_readme_names_only_atlas_techniques_read_in_the_latest_release():
    atlas = _entry("mitre-atlas")
    named = set(re.findall(r"AML\.T\d{4}(?:\.\d{3})?", README.read_text(encoding="utf-8")))
    assert named, "the README links no ATLAS technique; row 1 expects ATLAS to be wired in"
    unverified = named - set(atlas["relevant_entries"])
    assert not unverified, f"README names ATLAS techniques the register has not verified: {sorted(unverified)}"


def test_readme_names_the_atlas_release_the_register_calls_latest():
    text = README.read_text(encoding="utf-8")
    latest = _entry("mitre-atlas")["latest"]
    assert f"ATLAS {latest}" in text, f"README does not say 'ATLAS {latest}'"
    stale = set(re.findall(r"ATLAS (v\d{4}\.\d{2})", text)) - {latest}
    assert not stale, f"README names an ATLAS release other than the latest: {sorted(stale)}"


def test_readme_names_only_agentic_entries_read_at_the_publisher():
    agentic = _entry("owasp-agentic-top10")
    named = set(re.findall(r"\bASI\d{2}\b", README.read_text(encoding="utf-8")))
    assert "ASI06" in named, "the README does not link ASI06"
    unverified = named - set(agentic["relevant_entries"])
    assert not unverified, f"README names Agentic entries the register has not verified: {sorted(unverified)}"