llm-ingestion-okf/tests/test_excluded_terms.py

"""No tracked file carries a term from the locally kept exclusion list.

Sector-specific example material was replaced by generic, fictitious examples
written in this repository, and this test keeps it that way. Every tracked
file is read -- its PATH, its text, and for a binary fixture the text a reader
would get out of it (the members of a zip container, the inflated streams of a
PDF) -- and matched against the list.

**The list itself is not published.** It lives in a git-ignored file,
`tests/.excluded-terms.local.txt` (or the path in `OKF_EXCLUDED_TERMS`), one
regular expression per line, matched case-insensitively unless a line says
otherwise with an inline flag. Without the file the check is SKIPPED with a
message saying so -- never passed -- and the scanner's own known-positive runs
either way, on a synthetic term.

**The exception list is empty and is tested as a list:** an entry must name a
tracked file that still matches, so an exception can neither outlive its
reason nor be added for a path that does not exist.
"""

from __future__ import annotations

import os
import re
import subprocess
import zipfile
import zlib
from pathlib import Path

import pytest

PROJECT_ROOT = Path(__file__).resolve().parents[1]
TERMS_FILE = Path(
    os.environ.get("OKF_EXCLUDED_TERMS", PROJECT_ROOT / "tests" / ".excluded-terms.local.txt")
)

#: path -> the reason it may match. Empty, and a test keeps it honest.
EXCEPTIONS: dict[str, str] = {}

_PDF_STREAM = re.compile(rb"stream\r?\n(.*?)\r?\nendstream", re.DOTALL)


def compile_terms(lines: list[str]) -> re.Pattern[str]:
    terms = [line.strip() for line in lines]
    terms = [t for t in terms if t and not t.startswith("#")]
    assert terms, "the term list holds no term"
    return re.compile("|".join(f"(?:{t})" for t in terms), re.IGNORECASE)


def _terms() -> re.Pattern[str]:
    if not TERMS_FILE.is_file():
        pytest.skip(f"NOT CHECKED: no local term list at {TERMS_FILE}")
    return compile_terms(TERMS_FILE.read_text(encoding="utf-8").splitlines())


def _texts(path: Path) -> list[str]:
    """What a reader could get out of the file, as text to be matched."""
    data = path.read_bytes()
    texts = [data.decode("utf-8", errors="replace")]
    if zipfile.is_zipfile(path):
        with zipfile.ZipFile(path) as archive:
            for member in archive.namelist():
                texts.append(member)
                texts.append(archive.read(member).decode("utf-8", errors="replace"))
    if data.startswith(b"%PDF"):
        for match in _PDF_STREAM.finditer(data):
            try:
                texts.append(zlib.decompress(match.group(1)).decode("latin-1"))
            except zlib.error:
                continue
    return texts


def hits_in(pattern: re.Pattern[str], paths: list[str], root: Path) -> dict[str, list[str]]:
    """Every matched term, per path; a path with no match is absent."""
    found: dict[str, list[str]] = {}
    for rel in paths:
        terms = [m.group(0) for m in pattern.finditer(rel)]
        file = root / rel
        if file.is_file():
            for text in _texts(file):
                terms += [m.group(0) for m in pattern.finditer(text)]
        if terms:
            found[rel] = sorted({t.lower() for t in terms})
    return found


def _tracked() -> list[str]:
    try:
        out = subprocess.run(
            ["git", "ls-files", "-z"],
            cwd=PROJECT_ROOT,
            capture_output=True,
            check=True,
        ).stdout
    except (OSError, subprocess.CalledProcessError):
        pytest.skip("NOT CHECKED: not a git checkout, the tracked-file list cannot be read")
    return [p for p in out.decode("utf-8").split("\0") if p]


def test_the_scanner_finds_a_known_positive_in_every_form(tmp_path: Path) -> None:
    """Known-positive first: a scan that cannot find is not a scan."""
    pattern = compile_terms(["# a comment", "", r"\bzq\d{3}\b", "xyzzyterm"])
    (tmp_path / "plain.txt").write_text("see the XYZZYTERM handbook\n", encoding="utf-8")
    with zipfile.ZipFile(tmp_path / "container.docx", "w", zipfile.ZIP_DEFLATED) as archive:
        archive.writestr("word/document.xml", "<w:t>xyzzyterm</w:t>")
    stream = zlib.compress(b"BT (xyzzyterm) Tj ET")
    (tmp_path / "doc.pdf").write_bytes(b"%PDF-1.4\n1 0 obj\nstream\n" + stream + b"\nendstream\n")
    (tmp_path / "clean.txt").write_text("nothing to see\n", encoding="utf-8")

    found = hits_in(pattern, ["plain.txt", "container.docx", "doc.pdf", "clean.txt"], tmp_path)
    assert set(found) == {"plain.txt", "container.docx", "doc.pdf"}
    assert hits_in(pattern, ["zq500-x.md"], tmp_path) == {"zq500-x.md": ["zq500"]}


def test_the_term_list_is_not_tracked() -> None:
    assert "tests/.excluded-terms.local.txt" not in _tracked()


def test_every_exception_names_a_tracked_file_that_still_matches() -> None:
    pattern = _terms()
    tracked = set(_tracked())
    for path, reason in EXCEPTIONS.items():
        assert path in tracked, f"exception for an untracked path: {path}"
        assert reason.strip(), f"exception without a reason: {path}"
        assert hits_in(pattern, [path], PROJECT_ROOT), f"stale exception: {path}"


def test_no_tracked_file_carries_an_excluded_term() -> None:
    pattern = _terms()
    found = hits_in(pattern, [p for p in _tracked() if p not in EXCEPTIONS], PROJECT_ROOT)
    listing = "\n".join(f"  {path}: {', '.join(terms)}" for path, terms in sorted(found.items()))
    assert not found, f"{len(found)} tracked file(s) carry an excluded term:\n{listing}"