open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 4 Activity
llm-ingestion-okf/tests/test_accounting_gate.py
Kjell Tore Guttormsen bb8c926c5e
 test(accounting): the published tbx count is one number, guarded without the delivery 
`assert sum(tbx.values()) == 568` sat behind a `skipif` on a file only
this machine has, so on a fresh clone the sentence five files publish was
unguarded again -- the state in which 574 survived in four docstrings
until PM counted it.

`N101_TBX_TAGS` is now the one place the number lives, the delivery test
asserts against it, and a second test reads the published sentence out of
all five files and holds them to it. It needs no corpus and no clock:
editing CLAUDE.md to 600 is red on a fresh clone.

It was red at birth for a reason worth keeping: the scan read this test
file's own known-positive string (`574`) as a sixth publisher. The
known-positive is now assembled from pieces, and that failure is the
demonstration that the scan reads what it is pointed at.

What it does NOT prove is stated in the docstring: five files agreeing is
agreement, not a count. The measurement stays where it was.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-19 19:18:16 +02:00

1775 lines
72 KiB
Python
Raw Blame History

"""The content-accounting gate's own tests (`tools/okf_accounting_gate.py`).
The gate was written RED at `0b00de4`, before `okf build` had a source
inventory or a per-element account. These tests are GREEN and prove three things.
1. The fasit is independent. `tools/okf_witness.py` imports no
`llm_ingestion_okf` module -- checked on the live import graph of a process
that ran every witness, with a control that shows the check fires -- and
the committed inventories are exactly what the witness counts today, pinned
again to hand counts on four documents.
2. Every row CAN turn green and CAN turn red, each boundary driven from both
sides with synthetic build output (the door the capability must open).
3. Run against the real `okf build`, the fixture rows are green. They were
red at `0b00de4` on the three defects the gate was ordered for (no
inventory, a file carried AND rejected, a rejected document logged as "0
carried of 0 found"); `test_a_file_carried_through_a_document_and_rejected_
is_double_booked` and the row 4 tests keep those defects detectable.
"""
from __future__ import annotations
import json
import re
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any
import pytest
TOOLS = Path(__file__).resolve().parents[1] / "tools"
sys.path.insert(0, str(TOOLS))
import okf_accounting_gate as gate # noqa: E402
import okf_witness as witness # noqa: E402
TABLE = [
".csv",
".docx",
".htm",
".html",
".json",
".md",
".odt",
".pdf",
".pptx",
".rtf",
".txt",
".xlsx",
".xml",
]
# --- 1. the fasit ------------------------------------------------------------
_IMPORT_PROBE = """
import sys
sys.path.insert(0, {tools!r})
{preload}
import okf_witness as w
from pathlib import Path
fixtures = Path({fixtures!r})
w.witness_inbox(fixtures / "corpus")
w.witness_inbox(fixtures / "rejected")
w.count_sts_json((fixtures / "witness" / "prosess-84-sts.twin.json").read_bytes())
w.pdf_poppler(fixtures / "corpus" / "prosess-84-tabell.pdf")
print(sorted(m for m in sys.modules if m.split(".")[0] == "llm_ingestion_okf"))
"""
def _package_modules_after_witness(preload: str) -> list[str]:
script = _IMPORT_PROBE.format(tools=str(TOOLS), fixtures=str(gate.FIXTURES), preload=preload)
out = subprocess.run(
[sys.executable, "-c", script], capture_output=True, text=True, check=True
).stdout
result: list[str] = json.loads(out.strip().splitlines()[-1].replace("'", '"'))
return result
def test_the_witness_imports_no_module_of_the_package_it_judges() -> None:
pytest.importorskip("pdfplumber")
assert _package_modules_after_witness("") == []
def test_the_import_check_fires_when_the_package_is_loaded() -> None:
pytest.importorskip("pdfplumber")
assert "llm_ingestion_okf" in _package_modules_after_witness("import llm_ingestion_okf")
@pytest.mark.parametrize(
("inbox", "committed"),
[(gate.CORPUS, gate.INVENTORY), (gate.REJECTED, gate.REJECTED_INVENTORY)],
)
def test_the_committed_fasit_is_what_the_witness_counts(inbox: Path, committed: Path) -> None:
pytest.importorskip("pdfplumber")
assert witness.witness_inbox(inbox) == gate.load_inventory(committed)
def test_the_fixture_corpus_covers_every_readme_file_type() -> None:
inventory = gate.load_inventory(gate.INVENTORY)
assert gate.readme_types() == TABLE
assert {entry["suffix"] for entry in inventory["documents"].values()} == set(TABLE)
# Counted by hand from the fixture bytes, not by the witness. Seven of the
# thirteen documents were added 2026-09-18 because 20 of 63 element types had
# a count of ZERO in their only fixture, and a type that cannot appear cannot
# be lost visibly either (independent review, M-3).
HAND_COUNTS = {
"notat.md": {
"code_block": 1,
"heading": 2,
"image": 1,
"paragraph": 2,
"table": 1,
"table_row": 2,
},
"side.htm": {"cell": 4, "heading": 2, "image": 1, "list_item": 2, "paragraph": 1, "table": 1},
"figur.html": {
"cell": 4,
"heading": 1,
"image": 1,
"list_item": 2,
"paragraph": 1,
"table": 1,
},
"krav-rikt-tekstformat.rtf": {"cell": 56, "image": 0, "paragraph": 3, "table_row": 24},
"bilde.rtf": {"cell": 0, "image": 1, "paragraph": 2, "table_row": 0},
"prosess-84-sts.xml": {
"cell": 0,
"citation": 0,
"figure": 0,
"figure_caption": 0,
"footnote": 0,
"image": 2,
"list_item": 0,
"math": 0,
"paragraph": 2,
"section": 2,
"section_label": 2,
"table": 0,
"table_label": 0,
"title": 2,
},
"sts-rikt.xml": {
"cell": 4,
"citation": 1,
"figure": 1,
"figure_caption": 1,
"footnote": 1,
"image": 1,
"list_item": 1,
"math": 1,
"paragraph": 5,
"section": 1,
"section_label": 1,
"table": 1,
"table_label": 1,
"title": 1,
},
"topptekst-og-kommentar.docx": {
"cell": 2,
"comment": 1,
"endnote": 1,
"footnote": 1,
"header_footer": 2,
"heading": 1,
"image": 0,
"paragraph": 3,
"table": 1,
"text_box": 1,
},
"notater-og-skjult.pptx": {
"cell": 4,
"hidden_slide": 1,
"image": 0,
"note": 1,
"paragraph": 2,
"slide": 1,
"table": 2,
"title": 2,
},
"skjult-ark-og-formel.xlsx": {
"cell": 7,
"formula": 1,
"hidden_sheet": 1,
"image": 1,
"row": 3,
"sheet": 1,
},
"logg.txt": {"line": 4, "paragraph": 3},
"mengder.csv": {"cell": 6, "header_cell": 3, "row": 2},
"parametre.json": {"key": 6, "value": 6},
"prosess-84-tabell.pdf": {"image": 2, "page": 1},
"prosess-84-web.html": {
"cell": 0,
"heading": 1,
"image": 3,
"list_item": 0,
"paragraph": 3,
"table": 0,
},
"prosess-84-notat.docx": {
"cell": 0,
"comment": 0,
"endnote": 0,
"footnote": 0,
"header_footer": 0,
"heading": 0,
"image": 1,
"paragraph": 2,
"table": 0,
"text_box": 0,
},
"prosess-84-presentasjon.pptx": {
"cell": 0,
"hidden_slide": 0,
"image": 1,
"note": 0,
"paragraph": 0,
"slide": 1,
"table": 0,
"title": 1,
},
"prisark.xlsx": {
"cell": 18,
"formula": 0,
"hidden_sheet": 0,
"image": 0,
"row": 9,
"sheet": 2,
},
"krav-tekstdokument.odt": {
"annotation": 0,
"cell": 56,
"header_footer": 0,
"heading": 1,
"image": 0,
"list_item": 0,
"paragraph": 2,
"table": 2,
},
"liste-og-bilde.odt": {
"annotation": 1,
"cell": 2,
"header_footer": 2,
"heading": 1,
"image": 1,
"list_item": 2,
"paragraph": 4,
"table": 1,
},
}
def test_every_element_type_the_witness_counts_occurs_in_a_fixture() -> None:
"""M-3: 20 of 63 element types had a count of 0 in their only fixture, so
six of seven witness mutants survived -- a witness cannot be wrong about
something it never sees."""
pytest.importorskip("pdfplumber")
inventory = gate.load_inventory(gate.INVENTORY)
seen: dict[str, int] = {}
for entry in inventory["documents"].values():
for role, number in entry["elements"].items():
key = f"{entry['suffix']} {role}"
seen[key] = seen.get(key, 0) + number
absent = sorted(key for key, number in seen.items() if number == 0)
assert absent == [], f"element types with no occurrence in any fixture: {absent}"
def test_the_hand_counts_cover_every_document_of_the_corpus() -> None:
"""Row 1's fasit is the witness's own output, so a hand count is the only
thing in the loop that the witness did not produce. Four of thirteen
documents had one; every document has one now."""
inventory = gate.load_inventory(gate.INVENTORY)
assert sorted(HAND_COUNTS) == sorted(inventory["documents"])
@pytest.mark.parametrize("name", sorted(HAND_COUNTS))
def test_the_witness_matches_a_hand_count(name: str) -> None:
inventory = witness.witness_file(gate.CORPUS, gate.CORPUS / name)
assert inventory.elements == HAND_COUNTS[name]
def test_a_fenced_heading_is_not_a_heading_to_the_witness() -> None:
count, _ = witness.count_markdown("# Real\n\n```bash\n# not one\n```\n")
assert count.counts["heading"] == 1
assert count.counts["code_block"] == 1
assert count.texts["heading"] == [["# Real"]]
def test_the_sts_image_reference_resolves_through_the_graphics_directory() -> None:
inventory = witness.witness_file(gate.CORPUS, gate.STS_FIXTURE)
assert [ref.target for ref in inventory.images] == [
"graphics/tabell-84-2.png",
"graphics/figur-84-1.png",
]
def test_a_reference_above_the_document_resolves_to_nothing(tmp_path: Path) -> None:
(tmp_path / "secret.png").write_bytes(b"x")
(tmp_path / "docs").mkdir()
document = tmp_path / "docs" / "a.html"
assert witness.resolve_local(tmp_path, document, "../secret.png") is None
def test_the_witness_refuses_a_doctype() -> None:
with pytest.raises(witness.WitnessRefused):
witness.count_sts_xml(b'<!DOCTYPE x [<!ENTITY a "b">]><standard/>')
# --- M-1 / M-2: what the witness could not see -------------------------------
#
# Written RED 2026-09-18. Rows 2 and 3 require the build's inventory to EQUAL
# the witness's, so what the witness does not count, nothing can account for:
# an independent review put a header and a comment in a docx, measured 0 of
# either in the bundle, and the accounting still read "2 of 2 carried".
def test_the_docx_witness_counts_a_header_a_comment_and_a_text_box() -> None:
inventory = witness.witness_file(gate.CORPUS, gate.CORPUS / "topptekst-og-kommentar.docx")
for role in ("header_footer", "comment", "endnote", "text_box"):
assert inventory.elements[role] > 0, role
assert "Utkast" in " ".join(
piece for pieces in inventory.texts["header_footer"] for piece in pieces
)
def test_the_pptx_witness_counts_notes_and_does_not_call_a_hidden_slide_a_slide() -> None:
inventory = witness.witness_file(gate.CORPUS, gate.CORPUS / "notater-og-skjult.pptx")
assert inventory.elements["note"] > 0
assert inventory.elements["hidden_slide"] == 1
def test_the_xlsx_witness_counts_a_formula_and_a_hidden_sheet() -> None:
inventory = witness.witness_file(gate.CORPUS, gate.CORPUS / "skjult-ark-og-formel.xlsx")
assert inventory.elements["formula"] > 0
assert inventory.elements["hidden_sheet"] == 1
def test_the_odt_witness_counts_a_header_and_does_not_call_an_annotation_prose() -> None:
inventory = witness.witness_file(gate.CORPUS, gate.CORPUS / "liste-og-bilde.odt")
assert inventory.elements["header_footer"] > 0
assert inventory.elements["annotation"] == 1
def test_the_sts_witness_counts_citations_formulas_and_figure_captions() -> None:
inventory = witness.witness_file(gate.CORPUS, gate.CORPUS / "sts-rikt.xml")
for role in ("citation", "math", "figure", "figure_caption"):
assert inventory.elements[role] > 0, role
def test_the_json_role_map_reads_a_prefixed_tag_and_the_publishers_figcaption() -> None:
"""H3, measured by PM on N200 2026-09-18: the delivery carries 74
`mml:math` and 49 `figcaption`, and the witness counted 0 and 0.
`count_sts_xml` takes every tag through `_local`, which strips both a
Clark-notation namespace and a prefix; `count_sts_json` compared the RAW
string, so `mml:math` never reached `tag == "math"`. And the publisher's
JSON puts a figure's caption in `figcaption` under the `graphic`, not in
the `fig/caption` NISO-STS writes -- measured over the eight deliveries on
this machine: 430 `figcaption`, 0 of them under a `caption`.
The fixture is the JSON half of `sts-rikt.xml`'s family, and both halves
of the defect are in it."""
twin = gate.FIXTURES / "witness" / "sts-mml-and-figcaption.twin.json"
counts = witness.count_sts_json(twin.read_bytes()).counts
assert counts["math"] == 1
assert counts["figure_caption"] == 1
def test_a_prefixed_tag_that_names_no_role_stays_uncounted() -> None:
"""The known-negative for stripping the prefix: N101 ships 568 `tbx:`
tags, and not one of their local names is a role. A rule that generalises
is measured against what it must NOT take."""
for tag in ("tbx:term", "tbx:definition", "tbx:note", "tbx:termEntry"):
assert witness._sts_role_json(witness._local(tag), "sec", "body") is None
#: THE ONE PLACE THIS NUMBER LIVES. The count itself is measured over the
#: delivery by `test_n101s_own_prefixed_tags_are_counted_here_and_name_no_role`
#: -- but that test is `skipif`-gated on a file only this machine has, so on a
#: fresh clone the four published sentences were unguarded again, which is how
#: 574 survived in four docstrings until PM counted it. The guard below needs
#: no delivery: it reads the published sentences and holds them to each other
#: and to this constant.
N101_TBX_TAGS = 568
#: The published sentence, in the five files that carry it. Anchored on `N101`
#: and on the literal `tbx:` that follows the number, so it cannot match some
#: other count standing nearby.
_PUBLISHED_TBX = re.compile(r"N101(?:'s)?[^.]{0,80}?\*{0,2}(\d+)\*{0,2}\s*`tbx:`")
#: Every file that states it. A sentence moving to a sixth file without being
#: added here is the residual, and it is the same residual the constant
#: replaces one level down -- this list is what keeps the number readable in
#: one place rather than five.
_TBX_PUBLISHERS = (
"CHANGELOG.md",
"CLAUDE.md",
"tools/okf_witness.py",
"tests/test_accounting_gate.py",
"docs/2026-09-19-regnskapsgaten-rest-og-normaliseringsdoren.md",
)
def test_the_published_tbx_count_is_one_number_and_needs_no_delivery() -> None:
"""The published strings are held to each other, on any machine.
This is the half the measurement could not cover. `_tags_of` counts the
real delivery and is right to; it also cannot run where the delivery is
absent, and an assertion that skips guards nothing. Editing `CLAUDE.md` to
600 tomorrow is red here, on a fresh clone, with no corpus.
It proves nothing about the WORLD -- five files agreeing is agreement, not
a count -- which is why the delivery test keeps its own measurement and
this one only holds the sentences to the constant it asserts.
"""
root = Path(__file__).resolve().parents[1]
for name in _TBX_PUBLISHERS:
path = root / name
assert path.is_file(), f"{name}: the file that publishes the count is gone"
# Whitespace-folded first: the sentence wraps differently in each file.
text = " ".join(path.read_text(encoding="utf-8").split())
found = _PUBLISHED_TBX.findall(text)
assert found, f"{name}: the published sentence is gone, or no longer says `tbx:`"
assert [int(value) for value in found] == [N101_TBX_TAGS] * len(found), (
f"{name}: publishes {found}, and the number this repository stands behind "
f"is {N101_TBX_TAGS}"
)
# KNOWN-POSITIVE for the pattern itself: it must find a number that is
# NOT the published one, or the loop above could be passing over nothing.
# Assembled from pieces so the scan above does not read this line as a
# sixth publisher -- written whole, it made the guard red on its own
# fixture, which is also the clearest demonstration that the scan reads
# the file it is pointed at.
wrong = "N101 ships " + "574" + " `tbx:` tags"
assert _PUBLISHED_TBX.findall(wrong) == ["574"]
N101_DELIVERY = gate.N200_DEFAULT.parent / "N101-2025-860031.json"
def _tags_of(payload: bytes) -> dict[str, int]:
"""Every `tag` string in a delivery, counted by a walk written HERE.
The witness's own reader is what the known-negative below judges, so
counting through it would make the two agree by construction.
"""
names: dict[str, int] = {}
def walk(node: Any) -> None:
if isinstance(node, dict):
tag = node.get("tag")
if isinstance(tag, str):
names[tag] = names.get(tag, 0) + 1
for value in node.values():
walk(value)
elif isinstance(node, list):
for value in node:
walk(value)
walk(json.loads(payload.decode("utf-8")))
return names
@pytest.mark.skipif(not N101_DELIVERY.is_file(), reason="N101 is not on this machine")
def test_n101s_own_prefixed_tags_are_counted_here_and_name_no_role() -> None:
"""The published number for that known-negative was a measurement nothing
could falsify: it lived in four docstrings and in no assertion, and it was
wrong. The count is made HERE, over the delivery itself, so the sentence
four files publish is red when it stops being true."""
names = _tags_of(N101_DELIVERY.read_bytes())
assert sum(names.values()) > 0, "the walk found no tag at all"
tbx = {tag: n for tag, n in names.items() if tag.startswith("tbx:")}
assert sum(tbx.values()) == N101_TBX_TAGS
for tag in sorted(tbx):
assert witness._sts_role_json(witness._local(tag), "sec", "body") is None, tag
@pytest.mark.skipif(not gate.N200_DEFAULT.is_file(), reason="N200 is not on this machine")
def test_the_json_role_map_counts_n200s_own_formulas_and_figure_captions() -> None:
"""The same defect on the delivery it was found in, with PM's numbers.
Skipped where the corpus is absent, and then this file's own fixture is
the only thing holding the rule -- which is why both exist."""
counts = witness.count_sts_json(gate.N200_DEFAULT.read_bytes()).counts
assert counts["math"] == 74
assert counts["figure_caption"] == 49
assert counts["citation"] == 194
assert counts["figure"] == 49
def test_the_two_sts_role_maps_are_written_twice_and_not_shared() -> None:
"""M-2: both STS witnesses went through ONE `_sts_role`, so row 5 could
never see a hole in it. Two maps, each written for its own delivery."""
assert witness._sts_role_xml is not witness._sts_role_json
assert "_sts_role_json" not in witness._sts_role_xml.__code__.co_names
assert "_sts_role_xml" not in witness._sts_role_json.__code__.co_names
def test_every_witnessed_type_says_what_it_does_not_count() -> None:
assert set(witness.NOT_COUNTED) == set(witness.WITNESSED_SUFFIXES)
assert all(witness.NOT_COUNTED[suffix] for suffix in witness.WITNESSED_SUFFIXES)
def test_the_gate_prints_what_each_witness_does_not_count() -> None:
rendered = gate.render([])
assert "not counted" in rendered
for suffix in witness.WITNESSED_SUFFIXES:
assert suffix in rendered
# --- 2. every row can go both ways -------------------------------------------
#: The two headings of `a.md` as a bundle would carry them. The gate verifies
#: a booked `carried` against THIS, never against the declaration.
_BUNDLE_TEXT = (
"# Foerste overskrift\n\nProsa.\n\n"
"![](/assets/0123456789ab-x.png)\n_Source: x.png_\n\n# Andre overskrift\n"
)
def _inventory() -> dict[str, Any]:
return {
"documents": {
"a.md": {
"suffix": ".md",
"elements": {"heading": 2, "image": 1},
"texts": {
"heading": ["Foerste overskrift", "Andre overskrift"],
"image": [""],
},
"images": [{"kind": "local", "ref": "graphics/x.png", "target": "graphics/x.png"}],
},
},
"files": {"graphics/x.png": {"pointed_at_by": ["a.md"]}},
}
def _assets(*paths: Path) -> dict[str, str]:
"""The `assets/` directory a build wrote for these source files."""
return {f"{gate._sha12(p)}-{p.name}": gate._sha256(p) for p in paths}
def _build(
*,
accounting: dict[str, Any] | None = None,
sources: set[str] | None = None,
assets: dict[str, str] | None = None,
log: str = "",
exit_code: int = 0,
bundle_text: str = _BUNDLE_TEXT,
) -> gate.Build:
return gate.Build(
exit_code=exit_code,
log=log,
accounting=accounting,
source_files={"a.md"} if sources is None else sources,
assets=assets or {},
bundle_text=bundle_text,
)
def _ledger(*pairs: tuple[str, str]) -> dict[str, Any]:
"""The accounting a run writes when it REWROTE these pictures.
The gate's conversion route reads its pairs from here and confirms them
against the bundle text, so a test constructing a `Build` by hand has to
say what the run booked. Passing none of them is the case where the run
converted nothing -- which is every bundle before the viewable-asset
round, and the state a forged pointer block leaves the ledger in.
"""
return {
"accounting_version": 1,
"documents": [
{
"source_file": "a.md",
"status": "persisted",
"code": None,
"conversions": [{"from": before, "to": after} for before, after in pairs],
}
],
"files": [],
}
def test_the_judge_proves_a_CONVERTED_image_was_carried(tmp_path: Path) -> None:
"""A BMP reaches the bundle as a PNG, so the SOURCE's bytes are not in
`assets/` and never will be.
`asset_holds` proved a carry by hashing the source file and looking for
those bytes. That is right for an image carried verbatim and wrong for one
the build converts: the run did carry the picture, under a new digest the
bundle STATES, and a judge that only knew the first rule reported 19 of
R761's 50 images as claimed-and-not-found the day the conversion landed.
The second route is not the build's naming rule restated. The gate reads
the two digests the bundle writes, and then HASHES the asset itself: the
claim is only accepted when a file in `assets/` actually holds the bytes
the bundle says it wrote. A bundle claiming a conversion it did not
perform still fails.
"""
source = tmp_path / "figur.bmp"
source.write_bytes(b"BM" + b"\x00" * 200)
carried = tmp_path / "carried.png"
carried.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x01" * 64)
before = gate._sha256(source)
after = gate._sha256(carried)
text = (
f"![Figur](/assets/{after[:12]}-figur.png)\n"
f"Image: graphics/figur.bmp (8x4 px) -- converted from image/bmp "
f"sha256:{before} to image/png sha256:{after}\n"
)
ledger = _ledger((before, after))
build = _build(assets={f"{after[:12]}-figur.png": after}, bundle_text=text, accounting=ledger)
assert gate.asset_holds(build, source) is True
# Known-negative on the same shape: the run booked the conversion, and the
# file it names is not there. A route that read the claim alone would pass.
empty = _build(assets={}, bundle_text=text, accounting=ledger)
assert gate.asset_holds(empty, source) is False
# Known-negative two: an asset IS there, under a digest nothing tied to
# this source.
other = "0" * 64
stranger = _build(assets={f"{other[:12]}-x.png": other}, bundle_text=text, accounting=ledger)
assert gate.asset_holds(stranger, source) is False
# Known-negative three: the same bundle text, and a run that booked NO
# conversion. This is the pointer block a document can write for itself.
unbooked = _build(
assets={f"{after[:12]}-figur.png": after}, bundle_text=text, accounting=_ledger()
)
assert gate.asset_holds(unbooked, source) is False
# --- the claim the judge reads must be one the CODE wrote --------------------
def _png_bytes(pixel: bytes) -> bytes:
"""A real PNG, built with zlib alone -- no Pillow, no package import."""
import struct
import zlib
def chunk(kind: bytes, payload: bytes) -> bytes:
return (
len(payload).to_bytes(4, "big")
+ kind
+ payload
+ zlib.crc32(kind + payload).to_bytes(4, "big")
)
return (
b"\x89PNG\r\n\x1a\n"
+ chunk(b"IHDR", struct.pack(">IIBBBBB", 1, 1, 8, 2, 0, 0, 0))
+ chunk(b"IDAT", zlib.compress(b"\x00" + pixel, 9))
+ chunk(b"IEND", b"")
)
def _huge_bmp() -> bytes:
"""A 54-byte BMP declaring 50 000 x 50 000: refused `asset_too_large`,
never carried, and therefore a source no honest bundle can claim."""
import struct
dib = struct.pack("<IiiHHIIiiII", 40, 50_000, 50_000, 1, 8, 1, 0, 3779, 3779, 256, 256)
return b"BM" + struct.pack("<IHHI", 54, 0, 0, 54) + dib
def test_a_document_cannot_forge_a_conversion_claim(tmp_path: Path) -> None:
"""THE FASIT NEVER COMES FROM THE READER IT JUDGES -- including this route.
The conversion route used to read its two digests out of the bundle text.
Measured by PM 2026-09-19 that was a way IN for content this repository
does not trust: a BMP declaring 50 000 x 50 000, refused
`asset_too_large` and never carried, gave `asset_holds = True` from an
image's alt text and from ordinary body text. Anchoring the claim to a
pointer block closed those two and not the class -- a pointer block is two
lines of markdown, and one HTML file with two `<p>` elements writes them.
So the pair the judge believes comes from the RUN's accounting, and the
bundle text only confirms it. Every arm below is a way a document can put
the sentence, or the whole block, into a bundle; the run's ledger holds
ONE conversion and it is about other pictures, so the route is live and no
arm may reach it. The known-positive at the end books the pair and shows
the route still works.
Thirteen arms. Eleven are PM's own list of ways the shape anchoring
already refused, kept as a regression guard -- narrowing this rule later
must not quietly widen one of them -- and two are the form that defeated
it: a perfectly written pointer block the run never booked, once in the
bundle's ordinary text and once naming the asset it points at exactly.
"""
never_carried = tmp_path / "figur.bmp"
never_carried.write_bytes(_huge_bmp())
real = tmp_path / "ekte.png"
real.write_bytes(_png_bytes(b"\x10\x20\x30"))
before = gate._sha256(never_carried)
after = gate._sha256(real)
assets = {f"{after[:12]}-ekte.png": after}
clause = f"converted from image/bmp sha256:{before} to image/png sha256:{after}"
pointer = f"![Ekte](/assets/{after[:12]}-ekte.png)\nImage: ekte.png (1x1 px)"
# The same clause with U+00A0 where the spaces are: PM's N6.
nbsp_clause = clause.replace(" ", "\u00a0")
other = "0" * 64
arms = {
"plain body text": f"{pointer}\n\nProsess 84. {clause}. Se figuren over.\n",
"a table cell": f"{pointer}\n\n| Krav | Kilde |\n| --- | --- |\n| 84-1 | {clause} |\n",
"a figure caption": f"{pointer}\n\nFigur 84-1 -- {clause}\n",
"a fenced code block": f"{pointer}\n\n```\n{clause}\n```\n",
"link text with a URL": f"{pointer}\n\n[{clause}](https://example.invalid/x)\n",
"the clause with non-breaking spaces": f"{pointer}\n\n{nbsp_clause}\n",
"a clause naming another asset's digest": (
f"![Ekte](/assets/{after[:12]}-ekte.png)\n"
f"Image: ekte.png (1x1 px) -- converted from image/bmp sha256:{before} "
f"to image/png sha256:{other}\n"
),
"a pointer block naming another asset": (
f"![Figur](/assets/{other[:12]}-annen.png)\nImage: annen.png (1x1 px) -- {clause}\n"
f"{pointer}\n"
),
"an Image: line of its own": f"{pointer}\n\nImage: figur.bmp (8x4 px) -- {clause}\n",
"the whole block inside a code fence": (
f"{pointer}\n\n```\n![Figur](/assets/{after[:12]}-ekte.png)\n"
f"Image: figur.bmp (8x4 px) -- {clause}\n```\n"
),
"the whole block on one line": (
f"{pointer}\n\n![Figur](/assets/{after[:12]}-ekte.png) "
f"Image: figur.bmp (8x4 px) -- {clause}\n"
),
# The two that defeated the shape anchoring. Written exactly as the
# code writes them, because that is the point: the form is not a
# signature, and only the ledger can tell these from the real thing.
"a whole pointer block the run never booked": (
f"{pointer}\n\n![Figur](/assets/{after[:12]}-ekte.png)\n"
f"Image: figur.bmp (8x4 px) -- {clause}\n"
),
"that block alone in the bundle": (
f"![Figur](/assets/{after[:12]}-ekte.png)\nImage: figur.bmp (8x4 px) -- {clause}\n"
),
}
# The run booked ONE conversion and it is about neither of these files, so
# the route is live in every arm and nothing it could believe is true.
live = _ledger(("f" * 64, "e" * 64))
for label, text in arms.items():
build = _build(assets=assets, bundle_text=text, accounting=live)
assert gate.asset_holds(build, never_carried) is False, (
f"{label}: a document talked the judge into a carry that never happened"
)
# KNOWN-POSITIVE on the same bytes: the last arm's text, believed once the
# RUN books the pair. Without it every arm above would pass on a route
# that had simply stopped working.
honest = _build(
assets=assets,
bundle_text=arms["that block alone in the bundle"],
accounting=_ledger((before, after)),
)
assert gate.asset_holds(honest, never_carried) is True
def test_the_build_never_writes_a_claim_the_document_supplied(tmp_path: Path) -> None:
"""PM's measured path, end to end through the real `okf build`.
The judge reading only pointer blocks is half of it. The other half is
that an image's own LABEL is document text written INSIDE a pointer
block, so the door that puts it there must not let it emit the grammar
the judge reads. This builds the forgery PM measured: a BMP that is
refused and never carried, a real PNG that is, and an `alt` attribute
claiming the first became the second.
"""
pytest.importorskip("llm_ingestion_guard")
corpus = tmp_path / "inbox"
(corpus / "graphics").mkdir(parents=True)
never_carried = corpus / "graphics" / "figur.bmp"
never_carried.write_bytes(_huge_bmp())
real = corpus / "graphics" / "ekte.png"
real.write_bytes(_png_bytes(b"\x10\x20\x30"))
before = gate._sha256(never_carried)
after = gate._sha256(real)
clause = f"converted from image/bmp sha256:{before} to image/png sha256:{after}"
(corpus / "prosess.html").write_text(
"<!doctype html>\n<html><head><title>Prosess 84</title></head>\n<body>\n"
"<h1>84 Konstruksjoner av betong</h1>\n"
"<p>Toleranseklassene staar i figuren under.</p>\n"
f'<img src="graphics/ekte.png" alt="Figur 84-1 -- {clause}">\n'
"<p>Og den store figuren:</p>\n"
'<img src="graphics/figur.bmp" alt="Figur 84-2">\n'
f"<p>{clause}</p>\n"
"</body></html>\n",
encoding="utf-8",
)
build = gate.run_build(corpus, tmp_path / "work", door=False)
assert build.exit_code == 0, build.log
assert any(name.startswith(after[:12]) for name in build.assets), build.assets
assert not any(name.startswith(before[:12]) for name in build.assets), (
"the 50 000 x 50 000 BMP was carried; the arm measures nothing"
)
assert gate.asset_holds(build, never_carried) is False, (
"a document's own alt text talked the judge into a carry that never happened"
)
def _small_bmp() -> bytes:
"""An uncompressed 24-bit 2x2 BMP -- a real picture this build CONVERTS.
BMP is outside `VIEWABLE_MEDIA_TYPES`, so the run rewrites it as a PNG and
books the conversion. It is the known-positive every forgery arm below
needs: without it an arm could pass because the route stopped working.
"""
import struct
rows = (((255, 0, 0), (0, 255, 0)), ((0, 0, 255), (255, 255, 255)))
# A 24-bit row is padded to a multiple of four bytes; without the pad the
# reader refuses the file rather than guessing at the stride.
pad = b"\x00" * ((-2 * 3) % 4)
body = b"".join(b"".join(bytes((b, g, r)) for (r, g, b) in row) + pad for row in rows)
dib = struct.pack("<IiiHHIIiiII", 40, 2, 2, 1, 24, 0, len(body), 3779, 3779, 0, 0)
return b"BM" + struct.pack("<IHHI", 54 + len(body), 0, 0, 54) + dib + body
def _forgery_corpus(root: Path, body: str, note: str | None = None) -> tuple[Path, str, str]:
"""An inbox holding a refused BMP, a carried PNG, a CONVERTED BMP, and a
document that writes `body` -- whatever pointer-shaped text the arm tries.
Returns the inbox and the two digests the forgery ties together.
"""
inbox = root / "inbox"
(inbox / "graphics").mkdir(parents=True)
never_carried = inbox / "graphics" / "stor.bmp"
never_carried.write_bytes(_huge_bmp())
real = inbox / "graphics" / "ekte.png"
real.write_bytes(_png_bytes(b"\x10\x20\x30"))
(inbox / "graphics" / "figur.bmp").write_bytes(_small_bmp())
before = gate._sha256(never_carried)
after = gate._sha256(real)
(inbox / "prosess.html").write_text(
"<!doctype html>\n<html><head><title>Prosess 84</title></head>\n<body>\n"
"<h1>84 Konstruksjoner av betong</h1>\n"
'<img src="graphics/stor.bmp" alt="Stor">'
'<img src="graphics/ekte.png" alt="Ekte">'
'<img src="graphics/figur.bmp" alt="Figur">\n'
f"{body}\n</body></html>\n",
encoding="utf-8",
)
if note is not None:
(inbox / "notat.md").write_text(note, encoding="utf-8")
return inbox, before, after
def test_a_document_cannot_forge_the_WHOLE_pointer_block(tmp_path: Path) -> None:
"""THE FORM IS NOT A SIGNATURE -- measured by PM 2026-09-19 on `ae441ab`.
The previous round bound the conversion claim to a pointer block, which
closed the two routes PM had measured. It did not close the class: a
pointer block is two lines of markdown, and an ordinary HTML document
writes two lines of markdown by having two `<p>` elements. PM reproduced
it through the real `okf build` -- a BMP refused `asset_too_large` and
absent from `assets/` read as carried, from a document naming one digest
that is public in the bundle and one that is computable in advance.
So the claim is bound to what the RUN wrote: the build books each
conversion in its own accounting, which no document can reach, and the
bundle text is read only to CONFIRM. Each arm below is a whole build.
"""
pytest.importorskip("llm_ingestion_guard")
tail = "-- converted from image/bmp sha256:{before} to image/png sha256:{after}"
arms = {
"two <p> elements in one HTML file": (
"<p>![Fig](/assets/{after12}-ekte.png)</p>\n"
"<p>Image: graphics/ekte.png (1x1 px) " + tail + "</p>",
None,
),
"one <p> with a <br>": (
"<p>![Fig](/assets/{after12}-ekte.png)<br>"
"Image: graphics/ekte.png (1x1 px) " + tail + "</p>",
None,
),
"a markdown note beside the HTML carrier": (
"<p>Se notatet.</p>",
"# Notat\n\n![Fig](/assets/{after12}-ekte.png)\n"
"Image: graphics/ekte.png (1x1 px) " + tail + "\n",
),
}
for index, (label, (body, note)) in enumerate(arms.items()):
root = tmp_path / f"arm{index}"
root.mkdir()
# Two passes: the first learns the digests, the second writes the
# document that names them. The forger has the same information --
# both digests are readable from a bundle this build already wrote.
inbox, before, after = _forgery_corpus(root, "<p>placeholder</p>")
shaped = {"before": before, "after": after, "after12": after[:12]}
(inbox / "prosess.html").write_text(
(inbox / "prosess.html")
.read_text(encoding="utf-8")
.replace("<p>placeholder</p>", body.format(**shaped)),
encoding="utf-8",
)
if note is not None:
(inbox / "notat.md").write_text(note.format(**shaped), encoding="utf-8")
build = gate.run_build(inbox, root / "work", door=True)
assert build.exit_code == 0, build.log
never_carried = inbox / "graphics" / "stor.bmp"
assert not any(name.startswith(before[:12]) for name in build.assets), (
f"{label}: the refused BMP was carried; the arm measures nothing"
)
assert gate.asset_holds(build, never_carried) is False, (
f"{label}: a document wrote the pointer block and the judge believed it"
)
# KNOWN-POSITIVE on the same build: the picture the run really did
# convert still reads as held. Without it every arm above would pass
# on a route that had simply been switched off.
assert gate.asset_holds(build, inbox / "graphics" / "figur.bmp") is True, (
f"{label}: the run's own conversion stopped being provable"
)
def test_the_judge_proves_carriage_and_says_it_does_not_prove_fidelity(tmp_path: Path) -> None:
"""The limit, MEASURED here rather than trusted to the prose beside it.
PM's M10 2026-09-19: a mutated converter that writes a BLANK PNG gives
`asset_holds = True`. The bundle is internally consistent -- the digest it
claims to have written really is the asset's digest -- and the judge has
no opinion about whether those bytes hold the source's picture. The suite
fells that mutant (`test_the_carried_png_holds_the_source_pixels_exactly`
decodes both sides); this gate cannot, and the docstring's "a bundle
claiming a conversion it did not perform still fails" reads wider than
the route reaches.
So the limit is asserted in BOTH directions: it is real (the blank PNG is
accepted) and it is stated (the docstring names it). Teaching the judge
pixels is a different job; leaving a reader to infer the gap is not.
"""
source = tmp_path / "figur.bmp"
source.write_bytes(_huge_bmp())
blank = tmp_path / "blank.png"
blank.write_bytes(_png_bytes(b"\xff\xff\xff"))
before = gate._sha256(source)
after = gate._sha256(blank)
text = (
f"![Figur](/assets/{after[:12]}-figur.png)\n"
f"Image: figur.bmp (8x4 px) -- converted from image/bmp sha256:{before} "
f"to image/png sha256:{after}\n"
)
build = _build(
assets={f"{after[:12]}-figur.png": after},
bundle_text=text,
accounting=_ledger((before, after)),
)
assert gate.asset_holds(build, source) is True
doc = gate.asset_holds.__doc__ or ""
lowered = doc.lower()
assert "fidelity" in lowered, "the judge does not say what its second route cannot see"
assert "pixel" in lowered, "the limit is stated without naming what is not checked"
def _declared(heading: int = 2, image: int = 1, fate: str = "rejected") -> dict[str, Any]:
return {
"accounting_version": 1,
"documents": [
{
"source_file": "a.md",
"status": "persisted",
"code": None,
"inventory": {"heading": 2, "image": 1},
"fates": {
"heading": {"carried": heading},
"image": {"pointer": image},
},
}
],
"files": [{"source_file": "graphics/x.png", "fate": fate, "code": "extractor_unknown"}],
}
def _corpus(tmp_path: Path) -> Path:
(tmp_path / "graphics").mkdir()
(tmp_path / "graphics" / "x.png").write_bytes(b"png bytes")
return tmp_path
def test_row1_is_green_when_every_type_has_a_fasit() -> None:
inventory = {
"documents": {f"f{s}": {"suffix": s, "elements": {}} for s in TABLE},
"files": {},
}
row = gate.row1(TABLE, inventory, inventory)
assert (row.k, row.m, row.status) == (13, 13, gate.GREEN)
def test_row1_is_red_when_one_type_lacks_a_fasit() -> None:
inventory = {
"documents": {f"f{s}": {"suffix": s, "elements": {}} for s in TABLE[:-1]},
"files": {},
}
row = gate.row1(TABLE, inventory, inventory)
assert (row.k, row.m, row.status) == (12, 13, gate.RED)
def test_row1_is_red_when_the_committed_fasit_is_stale() -> None:
inventory = {"documents": {"f.md": {"suffix": ".md", "elements": {"heading": 1}}}}
fresh = {"documents": {"f.md": {"suffix": ".md", "elements": {"heading": 2}}}}
row = gate.row1([".md"], inventory, fresh)
assert (row.k, row.status) == (0, gate.RED)
def test_row2_is_red_without_the_door() -> None:
row = gate.row2([".md"], _inventory(), _build(), door=False)
assert (row.k, row.m, row.status) == (0, 1, gate.RED)
def test_row2_is_green_when_the_declared_inventory_equals_the_witness() -> None:
row = gate.row2([".md"], _inventory(), _build(accounting=_declared()), door=True)
assert (row.k, row.m, row.status) == (1, 1, gate.GREEN)
def test_row2_is_red_when_the_declared_inventory_is_one_off() -> None:
declared = _declared()
declared["documents"][0]["inventory"]["heading"] = 3
row = gate.row2([".md"], _inventory(), _build(accounting=declared), door=True)
assert (row.k, row.status) == (0, gate.RED)
def test_row3_is_green_when_every_element_and_file_has_one_fate(tmp_path: Path) -> None:
units = gate.account(_inventory(), _build(accounting=_declared()), _corpus(tmp_path))
row = gate.row3(units, door=True)
assert (row.k, row.m, row.status) == (2, 2, gate.GREEN)
@pytest.mark.parametrize(("heading", "u", "d"), [(1, 1, 0), (3, 0, 1)])
def test_row3_is_red_one_element_either_side(tmp_path: Path, heading: int, u: int, d: int) -> None:
units = gate.account(
_inventory(), _build(accounting=_declared(heading=heading)), _corpus(tmp_path)
)
document = units[0]
assert (document.unaccounted, document.double) == (u, d)
assert gate.row3(units, door=True).status == gate.RED
def test_row3_is_red_when_no_fate_is_declared(tmp_path: Path) -> None:
units = gate.account(_inventory(), _build(), _corpus(tmp_path))
assert units[0].unaccounted == 3
assert gate.row3(units, door=False).status == gate.RED
def test_row3_says_how_many_documents_were_refused_whole_and_what_they_cost() -> None:
"""P11 and P12, PM 2026-09-19: row 3's own summary of a refusal, and the
`refused=` column on its detail line, could each be deleted with this file
staying green at 106 passed. The LOSS is held -- `Unit.refused` keeps the
unit unclean and the note names the source and its code -- but what the
ROW says about it was decoration nothing pinned, and the row is what a
reader of the gate's output actually sees.
R, D and the element total are counted HERE, over the units this test
built, never read back off the row. The known-negative is the same units
without a refusal: the sentence has to change with them, or it is a
constant that happens to read true."""
refused_note = "refused whole: 17 element(s) declared rejected `fail_secure`"
units = [
gate.Unit("a.xml", "document", 0, 0, refused=17, notes=[refused_note]),
gate.Unit("b.json", "document", 0, 0, refused=4, notes=["refused whole: 4 element(s)"]),
gate.Unit("c.md", "document", 0, 0, verified=9),
gate.Unit("d.png", "file", 0, 0),
]
documents = [u for u in units if u.kind == "document"]
refused_docs = [u for u in documents if u.refused]
elements = sum(u.refused for u in units)
assert (len(refused_docs), len(documents), elements) == (2, 3, 21)
row = gate.row3(units, door=True)
assert row.status == gate.RED
assert (
f"{elements} element(s) lost with "
f"{len(refused_docs)} of {len(documents)} document(s) refused whole"
) in row.reason
for unit in refused_docs:
assert any(
unit.name in detail and f"refused={unit.refused}" in detail for detail in row.details
), unit.name
clean = [gate.Unit(u.name, u.kind, 0, 0, verified=9) for u in units]
assert (
"0 element(s) lost with 0 of 3 document(s) refused whole"
in gate.row3(clean, door=True).reason
)
def test_a_file_carried_through_a_document_and_rejected_is_double_booked(tmp_path: Path) -> None:
corpus = _corpus(tmp_path)
carried = _assets(corpus / "graphics" / "x.png")
units = gate.account(_inventory(), _build(accounting=_declared(), assets=carried), corpus)
assert (units[1].unaccounted, units[1].double) == (0, 1)
def test_a_file_carried_through_a_document_and_declared_carried_is_clean(tmp_path: Path) -> None:
corpus = _corpus(tmp_path)
carried = _assets(corpus / "graphics" / "x.png")
build = _build(accounting=_declared(fate="carried"), assets=carried)
assert gate.account(_inventory(), build, corpus)[1].clean
def test_a_file_declared_carried_without_its_bytes_is_unaccounted(tmp_path: Path) -> None:
build = _build(accounting=_declared(fate="carried"))
unit = gate.account(_inventory(), build, _corpus(tmp_path))[1]
assert (unit.unaccounted, unit.double) == (1, 0)
def test_an_unpointed_file_sharing_bytes_with_a_carried_one_is_not_carried(
tmp_path: Path,
) -> None:
corpus = _corpus(tmp_path)
(corpus / "graphics" / "twin.png").write_bytes(b"png bytes")
inventory = _inventory()
inventory["files"]["graphics/twin.png"] = {"pointed_at_by": []}
carried = _assets(corpus / "graphics" / "x.png")
units = gate.account(inventory, _build(assets=carried), corpus)
assert [(u.name, u.double) for u in units[1:]] == [
("graphics/twin.png", 0),
("graphics/x.png", 1),
]
def test_without_the_door_double_booking_is_derived_from_conservation(tmp_path: Path) -> None:
corpus = _corpus(tmp_path)
carried = _assets(corpus / "graphics" / "x.png")
assert gate.account(_inventory(), _build(assets=carried), corpus)[1].double == 1
assert gate.account(_inventory(), _build(), corpus)[1].clean
# --- B-1: the judge opens the bundle itself ----------------------------------
#
# Written RED 2026-09-18 against the hardening order. At 864570b the gate
# compared BOOKED NUMBERS with the witness's counts and never opened a concept
# file, so a report that booked every element of every document as carried was
# `GATE GREEN` over a bundle holding nothing (independent review, B-1).
def _all_carried(headings: int = 2, images: int = 1) -> dict[str, Any]:
"""A report that books everything as carried, the cheat's shape."""
return {
"accounting_version": 1,
"documents": [
{
"source_file": "a.md",
"status": "persisted",
"code": None,
"inventory": {"heading": 2, "image": 1},
"fates": {
"heading": {"carried": headings},
"image": {"carried": images},
},
}
],
"files": [
{"source_file": "graphics/x.png", "fate": "rejected", "code": "extractor_unknown"}
],
}
def test_carried_text_the_bundle_does_not_hold_is_unverified(tmp_path: Path) -> None:
build = _build(accounting=_all_carried(images=0), bundle_text="")
unit = gate.account(_inventory(), build, _corpus(tmp_path))[0]
assert unit.unverified == 2
assert not unit.clean
def test_carried_text_the_bundle_holds_verifies(tmp_path: Path) -> None:
corpus = _corpus(tmp_path)
build = _build(accounting=_all_carried(images=0), assets=_assets(corpus / "graphics" / "x.png"))
unit = gate.account(_inventory(), build, corpus)[0]
assert (unit.unverified, unit.verified) == (0, 2)
def test_one_heading_carried_of_two_in_the_bundle_is_unverified(tmp_path: Path) -> None:
build = _build(accounting=_all_carried(images=0), bundle_text="# Foerste overskrift\n")
unit = gate.account(_inventory(), build, _corpus(tmp_path))[0]
assert unit.unverified == 1
def test_an_image_booked_carried_without_its_bytes_is_unverified(tmp_path: Path) -> None:
"""The image element has no text of its own, so the only proof it was
carried is the asset. Without it the booking is not verifiable, and an
unverifiable booking is never clean."""
unit = gate.account(_inventory(), _build(accounting=_all_carried()), _corpus(tmp_path))[0]
assert unit.unverified >= 1
assert not unit.clean
def test_a_negative_booking_is_never_clean(tmp_path: Path) -> None:
declared = _all_carried()
declared["documents"][0]["fates"]["heading"] = {
"carried": 25,
"rejected": {"extractor_unknown": -15},
}
unit = gate.account(_inventory(), _build(accounting=declared), _corpus(tmp_path))[0]
assert unit.invalid >= 1
assert not unit.clean
# 25 + (-15) = 10 booked against a source holding 2, so eight are booked
# twice. Absorbing the sign would read 40 and report thirty-eight.
assert unit.double == 8
def test_a_document_declared_persisted_that_is_not_in_the_bundle_is_never_clean(
tmp_path: Path,
) -> None:
build = _build(accounting=_all_carried(), sources=set())
unit = gate.account(_inventory(), build, _corpus(tmp_path))[0]
assert unit.invalid >= 1
def test_everything_rejected_is_never_clean_for_a_document_the_build_persisted(
tmp_path: Path,
) -> None:
declared = _all_carried()
declared["documents"][0]["fates"] = {
"heading": {"rejected": {"fail_secure": 2}},
"image": {"rejected": {"fail_secure": 1}},
}
unit = gate.account(_inventory(), _build(accounting=declared), _corpus(tmp_path))[0]
assert unit.invalid >= 1
assert "persisted" in "; ".join(unit.notes)
def test_a_document_refused_whole_is_never_clean(tmp_path: Path) -> None:
"""H1, measured by PM 2026-09-18: a document the build refused books every
element as a coded rejection, so u = 0 and d = 0 and the unit read CLEAN.
The fate is honest and the content is gone; the gate has to say both."""
declared = _all_carried()
declared["documents"][0]["status"] = "rejected"
declared["documents"][0]["code"] = "fail_secure"
declared["documents"][0]["fates"] = {
"heading": {"rejected": {"fail_secure": 2}},
"image": {"rejected": {"fail_secure": 1}},
}
unit = gate.account(
_inventory(), _build(accounting=declared, sources=set()), _corpus(tmp_path)
)[0]
assert not unit.clean
assert unit.refused == 3
assert "fail_secure" in "; ".join(unit.notes)
def test_a_document_refused_whole_is_not_made_clean_by_the_neighbour(tmp_path: Path) -> None:
"""H1 in the shape the po scenario meets it: ONE refused source beside an
accepted one. `refused_whole` asks its question only when the corpus
persisted NOTHING, so the partial case reached row 3 as `clean = 4 of 4`
with `okf build` exiting 0 and three elements gone unseen."""
inventory = _inventory()
inventory["documents"]["b.md"] = {
"suffix": ".md",
"elements": {"heading": 1},
"texts": {"heading": [["Refused"]]},
"images": [],
}
declared = _all_carried()
declared["documents"].append(
{
"source_file": "b.md",
"status": "rejected",
"code": "fail_secure",
"inventory": {"heading": 1},
"fates": {"heading": {"rejected": {"fail_secure": 1}}},
}
)
declared["files"][0]["fate"] = "carried"
build = _build(
accounting=declared,
sources={"a.md"},
assets=_assets(_corpus(tmp_path) / "graphics" / "x.png"),
)
units = gate.account(inventory, build, tmp_path)
refused = next(u for u in units if u.name == "b.md")
assert gate.refused_whole(inventory["documents"], build) is None, "the corpus is not refused"
assert not refused.clean
assert (refused.name, refused.refused) == ("b.md", 1)
row = gate.row3(units, door=True)
assert row.status == gate.RED
assert any("b.md" in detail and "fail_secure" in detail for detail in row.details)
def test_a_document_declared_rejected_that_the_bundle_holds_is_never_clean(
tmp_path: Path,
) -> None:
"""H2 / mutant X2: the mirror of
`test_a_document_declared_persisted_that_is_not_in_the_bundle_is_never_clean`,
and the only one of B-1's six refusals no test drove. A report claiming a
document was refused while a concept in the bundle names it is the shape
that hides a persist gate that did not fire.
The known-negative on the same declaration: with no concept naming it,
the refusal is honest and the only finding is H1's own column."""
declared = _all_carried()
declared["documents"][0]["status"] = "rejected"
declared["documents"][0]["code"] = "fail_secure"
declared["files"][0]["fate"] = "carried"
corpus = _corpus(tmp_path)
assets = _assets(corpus / "graphics" / "x.png")
held = gate.account(
_inventory(), _build(accounting=declared, sources={"a.md"}, assets=assets), corpus
)[0]
assert held.invalid >= 1
assert "declared rejected" in "; ".join(held.notes)
honest = _all_carried()
honest["documents"][0]["status"] = "rejected"
honest["documents"][0]["code"] = "fail_secure"
honest["documents"][0]["fates"] = {
"heading": {"rejected": {"fail_secure": 2}},
"image": {"rejected": {"fail_secure": 1}},
}
honest["files"][0]["fate"] = "carried"
absent = gate.account(
_inventory(), _build(accounting=honest, sources=set(), assets=assets), corpus
)[0]
assert absent.invalid == 0
assert absent.refused == 3
def test_a_rejection_code_outside_the_closed_list_is_never_clean(tmp_path: Path) -> None:
declared = _all_carried()
declared["documents"][0]["status"] = "rejected"
declared["documents"][0]["code"] = "because_i_said_so"
declared["documents"][0]["fates"] = {
"heading": {"rejected": {"because_i_said_so": 2}},
"image": {"rejected": {"because_i_said_so": 1}},
}
unit = gate.account(
_inventory(), _build(accounting=declared, sources=set()), _corpus(tmp_path)
)[0]
assert unit.invalid >= 1
def test_an_accounting_version_the_gate_does_not_read_is_never_clean(tmp_path: Path) -> None:
declared = _all_carried()
declared["accounting_version"] = 2
units = gate.account(_inventory(), _build(accounting=declared), _corpus(tmp_path))
assert not any(u.clean for u in units)
def test_an_asset_with_the_right_name_and_the_wrong_bytes_is_not_carried(tmp_path: Path) -> None:
"""m-1: the check was a NAME check, so a zero-byte file called
`<sha12>-x.png` proved a carry."""
corpus = _corpus(tmp_path)
source = corpus / "graphics" / "x.png"
lying = {f"{gate._sha12(source)}-x.png": gate._sha256_bytes(b"")}
build = _build(accounting=_declared(fate="carried"), assets=lying)
assert gate.account(_inventory(), build, corpus)[1].unaccounted == 1
def test_an_asset_under_a_reduced_name_still_proves_the_carry(tmp_path: Path) -> None:
"""The build lowercases and folds the source's basename and sniffs the
suffix from the bytes. Measured on R761, a judge checking the FULL name
called 50 of 50 carried images missing -- the judge's defect, not the
build's, so the content address is the check and the readable tail is not.
"""
corpus = _corpus(tmp_path)
source = corpus / "graphics" / "x.png"
digest = gate._sha256(source)
reduced = {f"{digest[:12]}-25-0143-tabeller-r761-r762.jpeg": digest}
build = _build(accounting=_declared(fate="carried"), assets=reduced)
assert gate.account(_inventory(), build, corpus)[1].clean
def test_an_asset_holding_the_bytes_under_a_foreign_address_is_not_a_carry(
tmp_path: Path,
) -> None:
corpus = _corpus(tmp_path)
source = corpus / "graphics" / "x.png"
build = _build(accounting=_declared(fate="carried"), assets={"x.png": gate._sha256(source)})
assert gate.account(_inventory(), build, corpus)[1].unaccounted == 1
def test_the_cheat_that_books_everything_carried_makes_row3_red(tmp_path: Path) -> None:
"""The review's `MODE=carried`: a report that changes not one byte of the
bundle and books every element as carried."""
units = gate.account(
_inventory(), _build(accounting=_all_carried(), bundle_text=""), _corpus(tmp_path)
)
assert gate.row3(units, door=True).status == gate.RED
_HONEST_LOG = (
"* **Images**: 0 carried of 1 found, written to `assets/`.\n"
"* a.md: 3 elements found in the source, 0 carried: document rejected `fail_secure`\n"
)
def _rejected_inventory() -> dict[str, Any]:
inventory = _inventory()
inventory["documents"]["a.md"]["images"] = [{"kind": "local"}]
return inventory
def test_row4_is_green_when_the_log_names_what_the_rejected_document_held() -> None:
row = gate.row4(_rejected_inventory(), _build(sources=set(), log=_HONEST_LOG))
assert (row.k, row.m, row.status) == (1, 1, gate.GREEN)
@pytest.mark.parametrize(
"log",
[
_HONEST_LOG.replace("0 carried of 1 found", "0 carried of 0 found"),
_HONEST_LOG.replace("3 elements", "2 elements"),
_HONEST_LOG.replace(" `fail_secure`", ""),
],
)
def test_row4_is_red_when_the_log_understates_the_rejected_document(log: str) -> None:
row = gate.row4(_rejected_inventory(), _build(sources=set(), log=log))
assert (row.k, row.status) == (0, gate.RED)
def test_row4_cannot_be_green_when_nothing_was_rejected() -> None:
row = gate.row4(_rejected_inventory(), _build(log=_HONEST_LOG))
assert (row.m, row.status) == (0, gate.RED)
def test_row5_is_green_when_the_witnesses_agree() -> None:
row = gate.row5([("pair", gate.compare({"p": 3}, {"p": 3}))], [])
assert (row.k, row.m, row.status) == (1, 1, gate.GREEN)
def test_row5_is_red_with_both_numbers_when_they_disagree_by_one() -> None:
row = gate.row5([("pair", gate.compare({"p": 3}, {"p": 4}))], [])
assert row.status == gate.RED
assert row.details == ["pair: p: 3 vs 4"]
def test_row5_is_red_when_a_witness_is_missing() -> None:
assert gate.row5([("pair", gate.compare({"p": 3}, None))], []).status == gate.RED
def test_row6_without_its_source_is_red_locally_and_skipped_in_ci(tmp_path: Path) -> None:
missing = tmp_path / "absent"
local = gate.row6(missing, None, ci=False)
ci = gate.row6(missing, None, ci=True)
assert (local.status, local.fails) == (gate.RED, True)
assert (ci.status, ci.fails) == (gate.SKIPPED, False)
assert "source missing" in ci.reason
def test_a_skipped_row_never_leaves_the_verdict_unqualified() -> None:
"""m-2: `CI=1` with a missing source printed `GATE GREEN` with nothing
beside it, so the one line most readers stop at said the corpus passed."""
skipped = gate.Row(6, "real corpora", 0, 0, gate.SKIPPED, "not measured, source missing: x")
rendered = gate.render([skipped])
assert "GATE GREEN (row 6 not run: not measured, source missing: x)" in rendered
def test_a_corpus_refused_whole_under_the_default_gate_is_red(tmp_path: Path) -> None:
"""Row 6 was GREEN with R761 100 % rejected: every element booked as a
coded rejection satisfies u = 0 and d = 0. The build order asked for an
honest red there, so the row says this on its own."""
inventory = {
"documents": {
"a.md": {
"suffix": ".md",
"elements": {"heading": 2},
"texts": {"heading": [["A"], ["B"]]},
"images": [],
}
},
"files": {},
}
declared = {
"accounting_version": 1,
"documents": [
{
"source_file": "a.md",
"status": "rejected",
"code": "fail_secure",
"inventory": {"heading": 2},
"fates": {"heading": {"rejected": {"fail_secure": 2}}},
}
],
"files": [],
}
build = _build(accounting=declared, sources=set(), exit_code=1)
units = gate.account(inventory, build, tmp_path)
# The NUMBERS still balance -- that is what made the row green, and since
# H1 the loss has its own column instead of hiding behind them.
assert all((u.unaccounted, u.double) == (0, 0) for u in units)
assert not any(u.clean for u in units)
assert gate.refused_whole(inventory["documents"], build) is not None
def test_a_corpus_whose_every_document_has_no_declared_fate_says_so() -> None:
"""H6: N200 contributes one blank red. `okf build` proposes 0 plans on it
and FAILS (exit 2) before the accounting door is reached -- reproduced
2026-09-19: no accounting file is written at all -- so all 16 549 elements
land as `u` with `no declared fates` and the corpus measures none of the
classes it was brought in for. The row has to say that instead of showing
a number that looks like a finding about the build."""
blank = [
gate.Unit("a.json", "document", 9, 0, notes=["no declared fates"]),
gate.Unit("b.json", "document", 7, 0, notes=["no declared fates"]),
]
said = gate.measures_no_class(blank)
assert said is not None
assert "2 of 2" in said and "no declared fate" in said
def test_a_corpus_with_one_declared_document_is_not_called_blank() -> None:
"""The known-negative: one document with a fate is a corpus that measures
something, however badly the rest went."""
mixed = [
gate.Unit("a.json", "document", 9, 0, notes=["no declared fates"]),
gate.Unit("b.xml", "document", 0, 0, verified=3),
]
assert gate.measures_no_class(mixed) is None
assert gate.measures_no_class([]) is None
def test_the_two_real_corpora_are_named_and_the_second_is_not_r761() -> None:
"""R761 holds 0 `fig`, 0 formulas and 0 references, so the gate's only
real corpus could not see the hole in the STS role map."""
corpora = gate.real_corpora(Path("/r761"), Path("/n200.json"))
assert [c.label.split()[0] for c in corpora] == ["R761", "N200"]
def test_a_unit_clean_in_only_one_of_the_two_builds_is_not_clean() -> None:
"""M13: the two gates see different things, so either build could cover
for the other."""
clean = gate.Unit("a", "document", 0, 0)
dirty = gate.Unit("a", "document", 1, 0)
assert gate.clean_in_every_run([[clean], [clean]]) == 1
assert gate.clean_in_every_run([[clean], [dirty]]) == 0
def test_a_row_skipped_while_the_default_source_exists_exits_one(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""H5: the guard asked whether the corpora named by the ARGUMENTS are
available -- and row 6 is SKIPPED precisely when none of them is, so the
branch could never fire and no test covered it. The question it meant to
ask is about the machine: a corpus that is HERE and was pointed away from
is a row that did not run, and `CI=1` then printed a qualified GREEN and
exited 0.
Measured against its own known-negative below, so a guard that fires on
everything would not pass either."""
present = tmp_path / "corpus.json"
present.write_text("{}", encoding="utf-8")
skipped = gate.Row(6, "real corpora", 0, 0, gate.SKIPPED, "not measured, source missing: x")
monkeypatch.setattr(gate, "N200_DEFAULT", present)
monkeypatch.setattr(gate, "evaluate", lambda **kwargs: [skipped])
code = gate.main(["--r761", str(tmp_path / "absent"), "--n200", str(tmp_path / "absent.json")])
assert code == 1
assert "row 6 was skipped while its source exists" in capsys.readouterr().err
def test_a_row_skipped_with_no_source_on_the_machine_exits_zero(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The known-negative: nothing to measure is not a row that did not run."""
skipped = gate.Row(6, "real corpora", 0, 0, gate.SKIPPED, "not measured, source missing: x")
monkeypatch.setattr(gate, "R761_DEFAULT", tmp_path / "absent")
monkeypatch.setattr(gate, "N200_DEFAULT", tmp_path / "absent.json")
monkeypatch.setattr(gate, "evaluate", lambda **kwargs: [skipped])
assert gate.main([]) == 0
def test_a_surviving_mutant_is_not_exit_zero() -> None:
"""H4: the mutation harness returned `2 if errors else 0`, so a run that
printed `killed 0 of 1` and named the survivor exited 0. PM measured it on
a copy carrying only the X2 mutant. A harness nothing can fail is a report,
not a gate."""
import okf_gate_mutants as mutants
assert mutants.verdict(survived=[], errors=[]) == 0
assert mutants.verdict(survived=["X2"], errors=[]) == 1
assert mutants.verdict(survived=[], errors=["not applied"]) == 2
assert mutants.verdict(survived=["X2"], errors=["not applied"]) == 2
def test_the_gate_exits_one_when_a_row_is_red(monkeypatch: pytest.MonkeyPatch) -> None:
"""M14: nothing checked the real command's exit code, so `main` could
return 0 over a red table and no test would notice."""
pytest.importorskip("pdfplumber")
pytest.importorskip("pypandoc")
monkeypatch.delenv("CI", raising=False)
code = gate.main(["--r761", "/no/such/corpus", "--n200", "/no/such/file.json"])
assert code == 1
def test_a_file_with_no_declaration_is_unaccounted_when_conservation_failed(
tmp_path: Path,
) -> None:
"""M17: without the door, a file that is not merged counts as a coded
rejection ONLY because the build's own conservation identity held. A run
that exited non-zero has not earned that inference."""
corpus = _corpus(tmp_path)
failed = _build(exit_code=1, log="K1b FAILED")
assert gate.account(_inventory(), failed, corpus)[1].unaccounted == 1
assert gate.account(_inventory(), _build(), corpus)[1].clean
def test_a_merged_file_declared_carried_without_its_bytes_is_still_a_false_claim(
tmp_path: Path,
) -> None:
"""M05: with `fates` summing to one anyway, the false claim is the only
thing that can see it."""
corpus = _corpus(tmp_path)
build = _build(accounting=_declared(fate="carried"), sources={"a.md", "graphics/x.png"})
unit = gate.account(_inventory(), build, corpus)[1]
assert (unit.unaccounted, unit.clean) == (1, False)
def test_a_rejection_code_inside_an_element_fate_must_also_be_known(tmp_path: Path) -> None:
"""M24: the document's own `code` was checked and the per-element ones
were not, so a report could name any reason it liked for an element."""
declared = _all_carried()
declared["documents"][0]["fates"]["heading"] = {"rejected": {"because_i_said_so": 2}}
unit = gate.account(_inventory(), _build(accounting=declared), _corpus(tmp_path))[0]
assert unit.invalid >= 1
assert "because_i_said_so" in "; ".join(unit.notes)
def test_the_json_twin_is_read_with_the_json_role_map(tmp_path: Path) -> None:
"""M-2, measured on R761 2026-09-18: the XML delivery places a section's
label at `sec/label` (7 714, and 0 inside a title); the JSON delivery puts
2 760 of them inside the title. Read with the XML map, the twin loses
every one of those."""
twin = gate.FIXTURES / "witness" / "sts-label-in-title.twin.json"
assert witness.count_sts_json(twin.read_bytes()).counts["section_label"] == 1
assert witness._sts_role_xml("label", "title", "sec") is None
assert witness._sts_role_json("label", "title", "sec") == "section_label"
def test_an_approved_exception_is_read_and_says_what_it_does() -> None:
"""m-3: `APPROVED_EXCEPTIONS` was read by no row, so approving one changed
nothing and the list could have said anything."""
for suffix, element in gate.APPROVED_EXCEPTIONS:
assert "no denominator moves" in gate.exception_effect(suffix, element)
assert "WARNING" in gate.exception_effect(".pdf", "page")
assert "names nothing" in gate.exception_effect(".doc", "heading")
rendered = gate.render([])
assert "no denominator moves" in rendered
def test_the_gate_states_its_own_limits() -> None:
rendered = gate.render([])
assert "what this gate cannot check" in rendered
assert len(gate.LIMITS) >= 5
def test_every_witnessed_type_has_a_vocabulary() -> None:
assert set(gate.FORMAT_VOCABULARY) == set(witness.WITNESSED_SUFFIXES)
def test_the_proposed_exceptions_are_not_applied() -> None:
assert "NOT APPROVED" in gate.render([])
assert not {suffix for suffix, _ in gate.APPROVED_EXCEPTIONS} & {
item["suffix"] for item in gate.PROPOSED_EXCEPTIONS
}, "an exception cannot be both proposed and approved"
def test_the_operator_approved_the_pdf_exception_and_nothing_else() -> None:
"""Operator 2026-09-17, answering the gate's three proposals: the PDF one
only. It moves no number -- no witness counts a heading in a PDF -- so what
it changes is that the gap is a stated limit rather than an open question.
"""
assert gate.APPROVED_EXCEPTIONS == frozenset(
{(".pdf", "heading"), (".pdf", "paragraph"), (".pdf", "table")}
)
assert [item["suffix"] for item in gate.PROPOSED_EXCEPTIONS] == [
".xlsx",
".md .txt .csv .json .odt .rtf",
]
assert gate.APPROVED_ON in gate.render([])
def test_bad_usage_exits_two() -> None:
with pytest.raises(SystemExit) as exc:
gate.main(["--no-such-flag"])
assert exc.value.code == 2
# --- 3. the real build at this commit ----------------------------------------
@pytest.fixture(scope="module")
def real_rows() -> list[gate.Row]:
pytest.importorskip("pdfplumber")
pytest.importorskip("pypandoc")
return gate.evaluate(r761=None, n200=None, ci=True, consume=False)
def _cheating_report(inventory: dict[str, Any], mode: str) -> dict[str, Any]:
"""The review's `cheat.py`, as data: a report that changes not one byte of
the bundle and books every element as carried (or as rejected)."""
documents = []
for name, entry in inventory["documents"].items():
if mode == "carried":
fates = {kind: {"carried": n} for kind, n in entry["elements"].items()}
else:
fates = {
kind: {"rejected": {"extractor_unknown": n}} if n else {"rejected": {}}
for kind, n in entry["elements"].items()
}
documents.append(
{
"source_file": name,
"status": "persisted",
"code": None,
"inventory": dict(entry["elements"]),
"fates": fates,
}
)
files = [{"source_file": name, "fate": "carried", "code": None} for name in inventory["files"]]
return {"accounting_version": 1, "documents": documents, "files": files}
@pytest.mark.parametrize("mode", ["carried", "empty"])
def test_a_report_the_build_did_not_write_cannot_make_row3_green(mode: str) -> None:
"""B-1, end to end on the real fixture bundle. Until 2026-09-18 both modes
gave `GATE GREEN`, exit 0: the gate compared the report's numbers with the
witness's and never opened a concept file."""
pytest.importorskip("pdfplumber")
pytest.importorskip("pypandoc")
inventory = gate.load_inventory(gate.INVENTORY)
with tempfile.TemporaryDirectory() as tmp:
build = gate.run_build(gate.CORPUS, Path(tmp), door=True)
build.accounting = _cheating_report(inventory, mode)
units = gate.account(inventory, build, gate.CORPUS)
assert gate.row3(units, door=True).status == gate.RED
def test_the_door_exists() -> None:
assert gate.door_available()
def test_the_real_gate_names_what_the_build_does_not_account_for(
real_rows: list[gate.Row],
) -> None:
"""Rows 1-5 against the real `okf build`; row 6 needs R761 and is skipped
here.
Rows 2 and 3 were GREEN at `864570b` and are RED now, and that is the
hardening working rather than a regression: the witness counts thirteen
classes of content it could not see before, the build accounts for none
of them, and a number nobody counts is a loss nobody can report. The
named classes are the raw material for the next capability order.
"""
assert [(r.number, r.status) for r in real_rows] == [
(1, gate.GREEN),
(2, gate.RED),
(3, gate.RED),
(4, gate.GREEN),
(5, gate.GREEN),
(6, gate.SKIPPED),
]
unaccounted = " ".join(real_rows[2].details)
for element in (
"annotation",
"citation",
"comment",
"endnote",
"figure",
"figure_caption",
"formula",
"header_footer",
"hidden_sheet",
"hidden_slide",
"math",
"note",
"text_box",
):
assert f"{element}: 0 booked of" in unaccounted, element
# And no false red from the gate's own reading: every element the build
# DOES book as carried was found in the bundle.
assert "0 claimed and not found" in real_rows[2].details[0]
Reference in a new issue View git blame Copy permalink