"""Content accounting inside `okf build` -- the capability behind the gate. `tools/okf_accounting_gate.py` is the judge and was written red first. These tests pin what the gate cannot say on its own: the behaviour changes to the file-level identity and to the exit code, the per-format inventory against the independent witness, and that a real loss is FOUND rather than booked. """ from __future__ import annotations import contextlib import io import json import shutil import sys from pathlib import Path from typing import Any import pytest from llm_ingestion_okf import accounting, cli, corpus from llm_ingestion_okf.inbox import GateDecision FIXTURES = Path(__file__).parent / "fixtures" / "accounting" CORPUS = FIXTURES / "corpus" REJECTED = FIXTURES / "rejected" def _build(inbox: Path, tmp_path: Path, *extra: str, name: str = "bundle") -> tuple[int, Path, str]: bundle = tmp_path / name argv = [ "build", str(inbox), "--bundle", str(bundle), "--bundle-id", "acc", "--okf-version", "0.2", *extra, ] err = io.StringIO() with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(err): code = cli.main(argv) return code, bundle, err.getvalue() def _inbox(tmp_path: Path, source: Path) -> Path: target = tmp_path / "inbox" shutil.copytree(source, target) return target # --- d: a file carried through a document is not also a rejection ----------- def test_a_file_carried_through_a_document_is_not_a_coded_rejection(tmp_path: Path) -> None: pytest.importorskip("pypandoc") report = cli.build(CORPUS, tmp_path / "b", bundle_id="acc", okf_version="0.2") assert report.carried_files == 2 assert dict(report.codes).get("extractor_unknown", 0) == 0 assert report.merged + report.carried_files + report.rejected == report.n assert report.unaccounted == () def test_an_unpointed_image_beside_a_document_stays_a_coded_rejection(tmp_path: Path) -> None: pytest.importorskip("pypandoc") inbox = _inbox(tmp_path, CORPUS) shutil.copy(inbox / "graphics" / "figur-84-1.png", inbox / "graphics" / "ubrukt.png") report = cli.build(inbox, tmp_path / "b", bundle_id="acc", okf_version="0.2") assert report.carried_files == 2 assert dict(report.codes)["extractor_unknown"] == 1 def test_the_log_separates_carried_files_from_rejections(tmp_path: Path) -> None: pytest.importorskip("pypandoc") code, bundle, _ = _build(CORPUS, tmp_path) log = (bundle / "log.md").read_text(encoding="utf-8") assert code == 0 assert "merged + files carried through a document + coded rejections = 13 + 2 + 0 = 15" in log assert "`extractor_unknown`" not in log def test_a_corpus_without_carried_files_keeps_its_log_line(tmp_path: Path) -> None: inbox = tmp_path / "inbox" inbox.mkdir() (inbox / "a.md").write_text("# A\n\nText.\n", encoding="utf-8") code, bundle, _ = _build(inbox, tmp_path) log = (bundle / "log.md").read_text(encoding="utf-8") assert code == 0 assert "merged + coded rejections = 1 + 0 = 1; N = 1." in log # --- exit code: extracted but nothing persisted ------------------------------ def test_a_build_that_persisted_nothing_it_extracted_does_not_exit_zero(tmp_path: Path) -> None: code, _, err = _build(REJECTED, tmp_path) assert code == 1 assert "0 of 1 extracted document(s) persisted" in err def test_the_library_door_still_reports_all_rejected_without_raising(tmp_path: Path) -> None: report = corpus.measure( REJECTED, tmp_path / "b", ingested_at="1970-01-01T00:00:00Z", gate="guard-trusted-source" ) assert (report.extracted, report.persisted) == (1, 0) def test_a_folder_holding_no_document_still_exits_zero(tmp_path: Path) -> None: inbox = tmp_path / "inbox" inbox.mkdir() (inbox / "x.bin").write_bytes(b"x") code, _, _ = _build(inbox, tmp_path, "--segments", "off") assert code == 0 # --- the inventory equals the independent witness ----------------------------- def _witness(path: Path) -> dict[str, Any]: data: dict[str, Any] = json.loads(path.read_text(encoding="utf-8")) return data @pytest.mark.parametrize("name", sorted(_witness(FIXTURES / "inventory.json")["documents"])) def test_the_inventory_equals_the_witness(name: str) -> None: pytest.importorskip("pdfplumber") want = _witness(FIXTURES / "inventory.json")["documents"][name]["elements"] got = accounting.inventory(CORPUS, CORPUS / name) assert got.counts() == want def test_the_inventory_resolves_the_files_a_document_points_at() -> None: got = accounting.inventory(CORPUS, CORPUS / "prosess-84-sts.xml") assert got.pointed_files() == {"graphics/tabell-84-2.png", "graphics/figur-84-1.png"} def test_the_inventory_of_an_unread_type_is_empty() -> None: got = accounting.inventory(CORPUS, CORPUS / "graphics" / "figur-84-1.png") assert got.counts() == {} # --- the door ---------------------------------------------------------------- def test_the_door_books_every_element_of_the_fixture_corpus_once(tmp_path: Path) -> None: pytest.importorskip("pdfplumber") path = tmp_path / "accounting.json" code, _, err = _build(CORPUS, tmp_path, "--accounting", str(path)) assert code == 0, err data = json.loads(path.read_text(encoding="utf-8")) assert data["accounting_version"] == 1 assert data["unaccounted"] == 0 assert data["double_booked"] == 0 files = {entry["source_file"]: entry["fate"] for entry in data["files"]} assert files == {"graphics/figur-84-1.png": "carried", "graphics/tabell-84-2.png": "carried"} web = next(d for d in data["documents"] if d["source_file"] == "prosess-84-web.html") assert web["fates"]["image"] == {"carried": 2, "pointer": 1, "rejected": {}} def test_the_door_writes_the_accounting_into_the_log(tmp_path: Path) -> None: pytest.importorskip("pdfplumber") _, bundle, _ = _build(CORPUS, tmp_path, "--accounting", str(tmp_path / "a.json")) log = (bundle / "log.md").read_text(encoding="utf-8") assert "* **Accounting**: 13 document(s) and 2 other file(s);" in log assert "0 unaccounted, 0 double-booked." in log def test_a_rejected_document_is_logged_with_what_its_source_held(tmp_path: Path) -> None: code, bundle, _ = _build(REJECTED, tmp_path, "--accounting", str(tmp_path / "a.json")) log = (bundle / "log.md").read_text(encoding="utf-8") assert code == 1 assert ( "avvist.html: 3 elements found in the source, 0 carried: document rejected `fail_secure`" in log ) assert "**Images**: 0 carried of 1 found" in log def test_a_rejected_documents_elements_carry_its_code(tmp_path: Path) -> None: path = tmp_path / "a.json" _build(REJECTED, tmp_path, "--accounting", str(path)) data = json.loads(path.read_text(encoding="utf-8")) (doc,) = data["documents"] assert (doc["status"], doc["code"]) == ("rejected", "fail_secure") assert doc["fates"]["paragraph"] == {"carried": 0, "pointer": 0, "rejected": {"fail_secure": 1}} assert data["files"] == [ {"source_file": "graphics/figur.png", "fate": "rejected", "code": "extractor_unknown"} ] def test_a_lost_paragraph_is_found_and_fails_the_build( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """The known-positive: a gate that drops one line of text. Every file is still merged, K1b holds -- and the paragraph is gone.""" def dropping_gate(text: str) -> GateDecision: kept = "\n".join(line for line in text.split("\n") if "Vask skjer" not in line) return GateDecision(sanitized_text=kept, disposition="warn", reasons=()) monkeypatch.setattr(corpus, "resolve_gate", lambda name: dropping_gate) inbox = tmp_path / "inbox" inbox.mkdir() shutil.copy(CORPUS / "notat.md", inbox / "notat.md") path = tmp_path / "a.json" code, bundle, err = _build(inbox, tmp_path, "--accounting", str(path)) data = json.loads(path.read_text(encoding="utf-8")) assert code == 1 assert data["unaccounted"] == 1 (doc,) = data["documents"] assert doc["fates"]["paragraph"]["carried"] == 1 assert "accounting FAILED" in err assert "notat.md: paragraph 1 unaccounted" in (bundle / "log.md").read_text(encoding="utf-8") def test_without_the_door_the_log_has_no_accounting(tmp_path: Path) -> None: pytest.importorskip("pypandoc") _, bundle, _ = _build(CORPUS, tmp_path) assert "**Accounting**" not in (bundle / "log.md").read_text(encoding="utf-8") def test_the_door_is_a_build_flag(tmp_path: Path) -> None: parsed = cli.parse_args( ["build", "f", "--bundle", "b", "--okf-version", "0.2", "--accounting", "a.json"] ) assert parsed.accounting == Path("a.json") assert "accounting" in sys.modules["llm_ingestion_okf.accounting"].__name__ def test_a_proposed_exception_is_not_applied(tmp_path: Path) -> None: """An image inside a workbook is one of the three PROPOSED exceptions: the reader does not carry it. Until the operator approves the exception, it is unaccounted and the build says so -- it is never booked away silently.""" import zipfile inbox = tmp_path / "inbox" inbox.mkdir() with ( zipfile.ZipFile(Path(__file__).parent / "fixtures" / "prisark.xlsx") as source, zipfile.ZipFile(inbox / "bilde.xlsx", "w") as target, ): for name in source.namelist(): target.writestr(name, source.read(name)) target.writestr( "xl/drawings/drawing1.xml", '', ) pytest.importorskip("pypandoc") path = tmp_path / "a.json" code, bundle, _ = _build(inbox, tmp_path, "--accounting", str(path)) assert code == 1 assert json.loads(path.read_text(encoding="utf-8"))["unaccounted"] == 1 assert "bilde.xlsx: image 1 unaccounted." in (bundle / "log.md").read_text(encoding="utf-8") def test_a_converter_attribute_inside_carried_text_is_not_a_loss() -> None: """Measured on K2: the converter writes `\\[[Sted]{.mark}, [dd.mm.åååå]{.mark}\\]` for a highlighted `[Sted, dd.mm.åååå]`. The text is carried; the attribute letters between its words must not make it look lost.""" find = accounting._Finder("\\[[Sted]{.mark}, [dd.mm.åååå]{.mark}\\]\n") assert find("[Sted, dd.mm.åååå]") def test_a_line_break_inside_a_paragraph_splits_its_text(tmp_path: Path) -> None: """Measured on K2: `Ref.nr i tilbudet` in a table cell is written on two rows of a grid table, with other cells' text between the halves.""" import zipfile document = ( '' "Ref.nr i tilbudet" "" ) path = tmp_path / "a.docx" with zipfile.ZipFile(path, "w") as archive: archive.writestr("word/document.xml", document) inv = accounting.inventory(tmp_path, path) assert inv.counts()["paragraph"] == 1 assert inv.chunks == ["Ref.nr i ", "tilbudet"]