from __future__ import annotations
from copy import deepcopy
from io import BytesIO
from zipfile import ZipFile
import xml.etree.ElementTree as ET
from scrub_key import build_scrub_key
from scrub_key_document_reinsert import (
DOCX_LIMITATIONS,
reinsert_docx_bytes,
reinsert_text_document,
reinsert_txt_bytes,
)
VALID_SYNTHETIC_ROWS = [
{
"original_value": "BETROKKENE-TEST-A",
"placeholder": "[PERSOON_1]",
"entity_type": "PERSON",
"type_label": "Naam",
"source": "detected",
"review_status": "auto_detected",
"include": True,
"timestamp": "2026-06-08T10:00:00Z",
},
{
"original_value": "ZAAK-TEST-2026-001",
"placeholder": "[ZAAKNUMMER_1]",
"entity_type": "LEGAL_REFERENCE",
"type_label": "Zaaknummer",
"source": "manual",
"review_status": "manual",
"include": True,
"timestamp": "2026-06-08T10:05:00Z",
},
{
"original_value": "RECHTBANK TESTDAM",
"placeholder": "[ORGANISATIE_01]",
"entity_type": "ORGANIZATION",
"type_label": "Organisatie",
"source": "detected",
"review_status": "auto_detected",
"include": True,
"timestamp": "2026-06-08T10:10:00Z",
},
]
W_NS = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
def _valid_scrub_key():
return build_scrub_key(VALID_SYNTHETIC_ROWS, document_label="Document reinsert testdossier")
def _minimal_docx(document_xml: str) -> bytes:
content_types = """
"""
rels = """
"""
output = BytesIO()
with ZipFile(output, "w") as docx:
docx.writestr("[Content_Types].xml", content_types)
docx.writestr("_rels/.rels", rels)
docx.writestr("word/document.xml", document_xml)
return output.getvalue()
def _document_xml_with_texts(texts: list[str]) -> str:
paragraphs = "".join(
f"{text}" for text in texts
)
return f"""
{paragraphs}"""
def _document_xml_with_table(text: str) -> str:
return f"""
{text}"""
def _document_xml_with_split_run() -> str:
return f"""
[PERSOON_1]"""
def _docx_texts(docx_bytes: bytes) -> list[str]:
with ZipFile(BytesIO(docx_bytes), "r") as docx:
xml = docx.read("word/document.xml")
root = ET.fromstring(xml)
return [node.text or "" for node in root.iter(f"{{{W_NS}}}t")]
def test_txt_text_reinsert_with_one_placeholder():
result = reinsert_text_document("Verzoeker [PERSOON_1] verschijnt.", _valid_scrub_key())
assert result["text"] == "Verzoeker BETROKKENE-TEST-A verschijnt."
assert result["content"] == result["text"]
assert result["document_type"] == "txt"
assert result["replacement_count"] == 1
assert result["local_only"] is True
def test_txt_bytes_reinsert_with_multiple_placeholders():
content = "[PERSOON_1] heeft zaak [ZAAKNUMMER_1].".encode("utf-8")
result = reinsert_txt_bytes(content, _valid_scrub_key())
assert result["text"] == "BETROKKENE-TEST-A heeft zaak ZAAK-TEST-2026-001."
assert result["content_bytes"] == result["text"].encode("utf-8")
assert result["encoding"] == "utf-8"
assert result["replacement_count"] == 2
def test_txt_unknown_placeholder_remains_unchanged_and_is_reported():
result = reinsert_text_document("Onbekend: [ONBEKEND_1].", _valid_scrub_key())
assert result["text"] == "Onbekend: [ONBEKEND_1]."
assert result["unknown_placeholders"] == ["[ONBEKEND_1]"]
assert result["replacement_count"] == 0
def test_invalid_scrub_key_returns_validation_issues():
scrub_key = _valid_scrub_key()
scrub_key["items"][0]["timestamp"] = ""
result = reinsert_text_document("Waarde [PERSOON_1].", scrub_key)
assert result["text"] == "Waarde [PERSOON_1]."
assert result["replacement_count"] == 0
assert result["validation_issues"]
assert any("timestamp" in issue for issue in result["validation_issues"])
def test_docx_reinsert_with_one_placeholder_in_paragraph():
docx_bytes = _minimal_docx(_document_xml_with_texts(["Verzoeker [PERSOON_1] verschijnt."]))
result = reinsert_docx_bytes(docx_bytes, _valid_scrub_key())
assert result["document_type"] == "docx"
assert result["replacement_count"] == 1
assert _docx_texts(result["docx_bytes"]) == ["Verzoeker BETROKKENE-TEST-A verschijnt."]
def test_docx_reinsert_with_multiple_placeholders():
docx_bytes = _minimal_docx(
_document_xml_with_texts([
"[PERSOON_1] heeft zaak [ZAAKNUMMER_1].",
"Instantie: [ORGANISATIE_01].",
])
)
result = reinsert_docx_bytes(docx_bytes, _valid_scrub_key())
texts = _docx_texts(result["docx_bytes"])
assert texts == [
"BETROKKENE-TEST-A heeft zaak ZAAK-TEST-2026-001.",
"Instantie: RECHTBANK TESTDAM.",
]
assert result["replacement_count"] == 3
assert result["placeholders_not_found"] == []
def test_docx_reinsert_supports_table_text_in_main_document_xml():
docx_bytes = _minimal_docx(_document_xml_with_table("Tabelwaarde [ZAAKNUMMER_1]."))
result = reinsert_docx_bytes(docx_bytes, _valid_scrub_key())
assert _docx_texts(result["docx_bytes"]) == ["Tabelwaarde ZAAK-TEST-2026-001."]
assert result["replacement_count"] == 1
def test_docx_output_remains_a_valid_docx_zip_file():
docx_bytes = _minimal_docx(_document_xml_with_texts(["Waarde [PERSOON_1]."]))
result = reinsert_docx_bytes(docx_bytes, _valid_scrub_key())
with ZipFile(BytesIO(result["docx_bytes"]), "r") as docx:
assert "[Content_Types].xml" in docx.namelist()
assert "_rels/.rels" in docx.namelist()
assert "word/document.xml" in docx.namelist()
ET.fromstring(docx.read("word/document.xml"))
def test_docx_paragraph_text_is_restored_correctly():
docx_bytes = _minimal_docx(
_document_xml_with_texts(["Eerste alinea [PERSOON_1].", "Tweede alinea [ZAAKNUMMER_1]."])
)
result = reinsert_docx_bytes(docx_bytes, _valid_scrub_key())
assert _docx_texts(result["docx_bytes"]) == [
"Eerste alinea BETROKKENE-TEST-A.",
"Tweede alinea ZAAK-TEST-2026-001.",
]
def test_docx_helper_returns_audit_summary():
docx_bytes = _minimal_docx(_document_xml_with_texts(["Waarde [PERSOON_1]."]))
result = reinsert_docx_bytes(docx_bytes, _valid_scrub_key())
for field in [
"replacement_count",
"item_count",
"active_item_count",
"excluded_item_count",
"placeholders_not_found",
"unknown_placeholders",
"duplicate_placeholders",
"validation_issues",
"document_type",
"local_only",
"ai_processing",
"cloud_processing",
]:
assert field in result
assert result["document_type"] == "docx"
assert result["local_only"] is True
assert result["ai_processing"] is False
assert result["cloud_processing"] is False
def test_docx_unsupported_areas_or_limitations_are_documented():
docx_bytes = _minimal_docx(_document_xml_with_split_run())
result = reinsert_docx_bytes(docx_bytes, _valid_scrub_key())
assert result["limitations"] == DOCX_LIMITATIONS
assert result["unsupported_parts"] == DOCX_LIMITATIONS
assert any("split" in limitation.lower() for limitation in result["limitations"])
assert any("headers" in limitation.lower() for limitation in result["limitations"])
assert _docx_texts(result["docx_bytes"]) == ["[PERS", "OON_1]"]
assert result["replacement_count"] == 0
def test_helper_does_not_mutate_input_scrub_key():
scrub_key = _valid_scrub_key()
original = deepcopy(scrub_key)
docx_bytes = _minimal_docx(_document_xml_with_texts(["Waarde [PERSOON_1]."]))
reinsert_text_document("Waarde [PERSOON_1].", scrub_key)
reinsert_txt_bytes(b"Waarde [PERSOON_1].", scrub_key)
reinsert_docx_bytes(docx_bytes, scrub_key)
assert scrub_key == original
def test_no_ai_or_cloud_behavior():
txt_result = reinsert_text_document("Waarde [PERSOON_1].", _valid_scrub_key())
docx_bytes = _minimal_docx(_document_xml_with_texts(["Waarde [PERSOON_1]."]))
docx_result = reinsert_docx_bytes(docx_bytes, _valid_scrub_key())
for result in [txt_result, docx_result]:
assert result["local_only"] is True
assert result["ai_processing"] is False
assert result["cloud_processing"] is False
def test_examples_use_synthetic_values_only():
result = reinsert_text_document(
"[PERSOON_1] / [ZAAKNUMMER_1] / [ORGANISATIE_01]",
_valid_scrub_key(),
)
assert "BETROKKENE-TEST-A" in result["text"]
assert "ZAAK-TEST-2026-001" in result["text"]
assert "RECHTBANK TESTDAM" in result["text"]
assert "Jan Jansen" not in result["text"]
assert "Piet de Vries" not in result["text"]