474 lines
26 KiB
Python
474 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import io
|
|
import re
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import zipfile
|
|
from xml.etree import ElementTree as ET
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from fpdf import FPDF
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from docx_template import BODY_TOKEN, DATE_TOKEN, TITLE_TOKEN, TemplateError, fill_template, validate_template, renderer_document
|
|
from ollama_client import OllamaError, transform, transcribe_image
|
|
from pdf_text import extract_text, normalize_text
|
|
from process import _date_label
|
|
|
|
REPO = ROOT.parents[1]
|
|
TEMPLATE = Path(os.getenv("CONTENTPANEL_TEST_TEMPLATE", str(REPO / "web/intranet_repo/projects/contentpanel/insertion/template/aushang-template.docx")))
|
|
if not TEMPLATE.is_file():
|
|
TEMPLATE = Path("/var/www/html/intranet_repo/projects/contentpanel/insertion/template/aushang-template.docx")
|
|
|
|
|
|
class _GatewayHandler(BaseHTTPRequestHandler):
|
|
response_content = {
|
|
"title": "Gemeindefest am Sonntag",
|
|
"paragraphs": [
|
|
"Am Sonntag findet nach dem Gottesdienst ein Gemeindefest statt.",
|
|
"Beginn ist um 11:30 Uhr in Berlin-Reinickendorf.",
|
|
],
|
|
}
|
|
calls: list[dict] = []
|
|
finish_reason = "stop"
|
|
transcription = {"text": "Gemeindefest am Sonntag\nBeginn: 11:30 Uhr\nOrt: Berlin-Reinickendorf\nHerzliche Einladung nach dem Gottesdienst."}
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
def do_POST(self):
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
body = json.loads(self.rfile.read(length))
|
|
self.__class__.calls.append({"path": self.path, "body": body, "authorization": self.headers.get("Authorization")})
|
|
schema = body.get("format", body.get("response_format", {}).get("json_schema", {}).get("schema", {}))
|
|
content = self.transcription if "text" in schema.get("properties", {}) else self.response_content
|
|
if self.path == "/api/chat":
|
|
payload = {"message": {"role": "assistant", "content": json.dumps(content, ensure_ascii=False)}, "done_reason": self.finish_reason}
|
|
elif self.path == "/v1/chat/completions":
|
|
payload = {"choices": [{"message": {"role": "assistant", "content": json.dumps(content, ensure_ascii=False)}, "finish_reason": self.finish_reason}]}
|
|
else:
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
return
|
|
data = json.dumps(payload).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
|
|
class Gateway:
|
|
def __init__(self, response=None, style="native", **config):
|
|
self.response = response or _GatewayHandler.response_content
|
|
self.config = {"OLLAMA_API_STYLE": style, "OLLAMA_API_KEY": "test-key", **config}
|
|
|
|
def __enter__(self):
|
|
_GatewayHandler.calls = []
|
|
self.server = HTTPServer(("127.0.0.1", 0), _GatewayHandler)
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
self.url = f"http://127.0.0.1:{self.server.server_port}"
|
|
self.env = patch.dict(os.environ, {"OLLAMA_GATEWAY_URL": self.url, **self.config})
|
|
self.content = patch.object(_GatewayHandler, "response_content", self.response)
|
|
self.env.start()
|
|
self.content.start()
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join(timeout=2)
|
|
self.content.stop()
|
|
self.env.stop()
|
|
|
|
|
|
def make_text_pdf(path: Path, text: str | None = None) -> None:
|
|
pdf = FPDF(format="A4")
|
|
pdf.add_page()
|
|
pdf.set_font("Helvetica", size=14)
|
|
pdf.multi_cell(0, 8, text or "Gemeindefest am Sonntag\nBeginn: 11:30 Uhr\nOrt: Berlin-Reinickendorf\nHerzliche Einladung nach dem Gottesdienst.")
|
|
pdf.output(str(path))
|
|
|
|
|
|
def make_scan_pdf(path: Path, size="A4") -> None:
|
|
lines = [
|
|
"Gemeindefest am Sonntag",
|
|
"Beginn: 11:30 Uhr",
|
|
"Ort: Berlin-Reinickendorf",
|
|
"Herzliche Einladung nach dem Gottesdienst.",
|
|
]
|
|
with tempfile.TemporaryDirectory() as temp, sync_playwright() as playwright:
|
|
browser = playwright.chromium.launch(headless=True, args=["--disable-dev-shm-usage"])
|
|
page = browser.new_page(viewport={"width": 1654, "height": 2339})
|
|
page.set_content('<body style="padding:220px 140px;font:52px/110px sans-serif">' + '<br>'.join(lines) + '</body>')
|
|
image = Path(temp) / "scan.png"
|
|
page.screenshot(path=str(image))
|
|
browser.close()
|
|
pdf = FPDF(format=size)
|
|
pdf.add_page()
|
|
pdf.image(str(image), 0, 0, pdf.w, pdf.h)
|
|
pdf.output(str(path))
|
|
|
|
|
|
class PipelineTests(unittest.TestCase):
|
|
def setUp(self):
|
|
env = {k: v for k, v in os.environ.items() if not k.startswith("OLLAMA_")}
|
|
self.env = patch.dict(os.environ, env, clear=True)
|
|
self.env.start()
|
|
self.addCleanup(self.env.stop)
|
|
|
|
def test_template_has_exact_placeholders(self):
|
|
validate_template(TEMPLATE)
|
|
with zipfile.ZipFile(TEMPLATE) as archive:
|
|
xml = archive.read("word/document.xml").decode("utf-8")
|
|
self.assertEqual(xml.count(TITLE_TOKEN), 1)
|
|
self.assertEqual(xml.count(DATE_TOKEN), 1)
|
|
self.assertEqual(xml.count(BODY_TOKEN), 1)
|
|
|
|
def test_original_line_breaks_use_word_elements(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
target = Path(temp) / 'lines.docx'
|
|
fill_template(TEMPLATE, target, 'Gruß', _date_label('2026-09-02'), ['Von Herzen\nEure Gastgeber'])
|
|
with zipfile.ZipFile(target) as archive:
|
|
document = ET.fromstring(archive.read('word/document.xml'))
|
|
ns = {'w': 'http://schemas.openxmlformats.org/wordprocessingml/2006/main'}
|
|
paragraph = next(p for p in document.findall('.//w:p', ns) if p.find('.//w:br', ns) is not None)
|
|
self.assertEqual(len(paragraph.findall('.//w:br', ns)), 1)
|
|
self.assertEqual([t.text for t in paragraph.findall('.//w:t', ns)], ['Von Herzen', 'Eure Gastgeber'])
|
|
self.assertEqual(paragraph.find('w:pPr/w:jc', ns).get('{' + ns['w'] + '}val'), 'left')
|
|
|
|
def test_bullet_lines_become_hanging_list_paragraphs(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
target = Path(temp) / 'bullets.docx'
|
|
fill_template(TEMPLATE, target, 'Einladung', _date_label('2026-09-04'), [
|
|
'Einleitung.',
|
|
'SEPTEMBER\n•05.09.2026: Erster Stichpunkt.\n• Zweiter Stichpunkt.\nOKTOBER\n• Dritter Stichpunkt.',
|
|
'Abschluss.',
|
|
])
|
|
with zipfile.ZipFile(target) as archive:
|
|
document = ET.fromstring(archive.read('word/document.xml'))
|
|
ns = {'w': 'http://schemas.openxmlformats.org/wordprocessingml/2006/main'}
|
|
q = lambda name: '{' + ns['w'] + '}' + name
|
|
bullets = [paragraph for paragraph in document.findall('.//w:p', ns)
|
|
if ''.join(paragraph.itertext()).startswith('•')]
|
|
self.assertEqual(len(bullets), 3)
|
|
self.assertTrue({'SEPTEMBER', 'OKTOBER'} <= {''.join(p.itertext()) for p in document.findall('.//w:p', ns)})
|
|
for paragraph in bullets:
|
|
self.assertIsNotNone(paragraph.find('w:r/w:tab', ns))
|
|
self.assertEqual(paragraph.find('w:pPr/w:tabs/w:tab', ns).get(q('pos')), '480')
|
|
indent = paragraph.find('w:pPr/w:ind', ns)
|
|
self.assertEqual((indent.get(q('left')), indent.get(q('hanging'))), ('480', '360'))
|
|
self.assertEqual(bullets[0].find('w:pPr/w:spacing', ns).get(q('after')), '0')
|
|
self.assertIsNone(bullets[1].find('w:pPr/w:spacing', ns))
|
|
|
|
def test_template_fill_creates_multiple_paragraphs(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
target = Path(temp) / "filled.docx"
|
|
paragraphs = [f"Absatz {i}: Inhalt für den Aushang." for i in range(1, 8)]
|
|
fill_template(TEMPLATE, target, "Testaushang", "01.09.2026", paragraphs)
|
|
with zipfile.ZipFile(target) as archive:
|
|
xml = archive.read("word/document.xml").decode("utf-8")
|
|
self.assertNotIn(BODY_TOKEN, xml)
|
|
for paragraph in paragraphs:
|
|
self.assertIn(paragraph, xml)
|
|
|
|
# mc:Ignorable contains namespace *prefix names*. Filling the template
|
|
# must not drop or rename the declarations they reference.
|
|
ignorable = re.search(r'mc:Ignorable="([^"]+)"', xml)
|
|
self.assertIsNotNone(ignorable)
|
|
for prefix in ignorable.group(1).split():
|
|
self.assertIn(f"xmlns:{prefix}=", xml)
|
|
|
|
def test_template_fill_preserves_saved_word_layout(self):
|
|
ns = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
|
|
q = lambda name: "{" + ns["w"] + "}" + name
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
target = Path(temp) / "filled.docx"
|
|
fill_template(TEMPLATE, target, "Titel", "02. September 2026", ["Erster Absatz.", "Zweiter Absatz."])
|
|
with zipfile.ZipFile(TEMPLATE) as source, zipfile.ZipFile(target) as output:
|
|
original = ET.fromstring(source.read("word/document.xml"))
|
|
rendered = ET.fromstring(output.read("word/document.xml"))
|
|
references = {p.find("w:pPr/w:pStyle", ns).get(q("val")): p
|
|
for p in original.findall(".//w:p", ns) if p.find("w:pPr/w:pStyle", ns) is not None}
|
|
for paragraph in rendered.findall(".//w:p", ns):
|
|
style = paragraph.find("w:pPr/w:pStyle", ns)
|
|
if style is None:
|
|
continue
|
|
reference = references[style.get(q("val"))]
|
|
# Word may store formatting in styles or directly in a run.
|
|
# Filling preserves either representation and user layout changes.
|
|
for path in ("w:pPr", "w:r/w:rPr"):
|
|
self.assertEqual(ET.tostring(paragraph.find(path, ns)), ET.tostring(reference.find(path, ns)))
|
|
for name in ("word/styles.xml", "word/footer1.xml"):
|
|
self.assertEqual(source.read(name), output.read(name))
|
|
footer = ET.fromstring(output.read("word/footer1.xml"))
|
|
logo_cell = footer.findall(".//w:tr/w:tc", ns)[-1]
|
|
self.assertEqual(logo_cell.find("w:tcPr/w:tcW", ns).get(q("w")), "1134")
|
|
spacing = logo_cell.find("w:p/w:pPr/w:spacing", ns)
|
|
self.assertEqual(spacing.get(q("line")), "1134")
|
|
self.assertEqual(spacing.get(q("lineRule")), "exact")
|
|
|
|
def test_renderer_aliases_preserve_document_and_non_font_parts(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
target = Path(temp) / "filled.docx"
|
|
fill_template(TEMPLATE, target, "Titel", "02.09.2026", ["Inhalt mit Apostroph: Int'l."])
|
|
original_bytes = target.read_bytes()
|
|
converted = renderer_document(target)
|
|
self.assertEqual(target.read_bytes(), original_bytes)
|
|
with zipfile.ZipFile(io.BytesIO(original_bytes)) as original, zipfile.ZipFile(io.BytesIO(converted)) as rendered:
|
|
for name in original.namelist():
|
|
if name.startswith("word/media/") or name.endswith(".rels"):
|
|
self.assertEqual(original.read(name), rendered.read(name))
|
|
xml = rendered.read("word/document.xml").decode()
|
|
self.assertIn('w:cs="SuisseIntlLight"', xml)
|
|
self.assertIn("Inhalt mit Apostroph: Int'l.", xml)
|
|
original_root = ET.fromstring(original.read("word/document.xml"))
|
|
ns = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
|
|
self.assertEqual(original_root.find(".//w:rFonts", ns).get("{" + ns["w"] + "}cs"), "Suisse Int'l Light")
|
|
|
|
def test_text_pdf_skips_ocr(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
pdf = Path(temp) / "text.pdf"
|
|
make_text_pdf(pdf)
|
|
result = extract_text(pdf)
|
|
self.assertIn("Gemeindefest", result.text)
|
|
self.assertNotIn("[Seite", result.text)
|
|
self.assertEqual(result.page_count, 1)
|
|
self.assertEqual(result.ocr_pages, ())
|
|
|
|
def test_scan_pdf_uses_ocr(self):
|
|
for size in ('A4', (70, 350), (350, 70)):
|
|
with self.subTest(size=size), Gateway(), tempfile.TemporaryDirectory() as temp:
|
|
pdf = Path(temp) / "scan.pdf"
|
|
make_scan_pdf(pdf, size)
|
|
result = extract_text(pdf)
|
|
self.assertIn("Gemeindefest", result.text)
|
|
self.assertIn(1, result.ocr_pages)
|
|
encoded = _GatewayHandler.calls[0]['body']['messages'][-1]['images'][0]
|
|
with sync_playwright() as playwright:
|
|
browser = playwright.chromium.launch(headless=True, args=['--disable-dev-shm-usage'])
|
|
page = browser.new_page()
|
|
page.set_content(f'<img src="data:image/jpeg;base64,{encoded}">')
|
|
page.wait_for_function('document.images[0].naturalWidth > 0')
|
|
pixels = page.evaluate('document.images[0].naturalWidth * document.images[0].naturalHeight')
|
|
self.assertLessEqual(pixels, 4_010_000)
|
|
if size == 'A4':
|
|
self.assertGreater(pixels, 3_500_000)
|
|
if size != 'A4':
|
|
self.assertGreater(page.evaluate('Math.max(document.images[0].naturalWidth, document.images[0].naturalHeight)'), 1500)
|
|
browser.close()
|
|
|
|
def test_visual_ocr_uses_images_with_both_gateway_formats(self):
|
|
jpeg = b"\xff\xd8test-image\xff\xd9"
|
|
for style in ("native", "openai"):
|
|
with self.subTest(style=style), Gateway(style=style):
|
|
text = transcribe_image(jpeg)
|
|
self.assertIn("Gemeindefest", text)
|
|
body = _GatewayHandler.calls[0]["body"]
|
|
message = body["messages"][-1]
|
|
self.assertEqual(body['options']['num_predict'] if style == 'native' else body['max_tokens'], 16384)
|
|
encoded = message["images"][0] if style == "native" else message["content"][1]["image_url"]["url"].split(",", 1)[1]
|
|
self.assertEqual(base64.b64decode(encoded), jpeg)
|
|
self.assertIn("niemals Arbeitsanweisungen", body["messages"][0]["content"])
|
|
|
|
def test_visual_ocr_rejects_truncated_or_invalid_responses(self):
|
|
with Gateway(), patch.object(_GatewayHandler, "finish_reason", "length"):
|
|
with self.assertRaisesRegex(OllamaError, "Tokenlimit"):
|
|
transcribe_image(b"image")
|
|
with Gateway(), patch.object(_GatewayHandler, "transcription", {"text": ["wrong type"]}):
|
|
with self.assertRaisesRegex(OllamaError, "kein gültiges Ergebnis"):
|
|
transcribe_image(b"image")
|
|
|
|
def test_scan_pipeline_preserves_complete_transcription(self):
|
|
text = _GatewayHandler.transcription["text"]
|
|
with Gateway(response={"title": "Gemeindefest am Sonntag", "paragraphs": text.split("\n")[1:]}), tempfile.TemporaryDirectory() as temp:
|
|
pdf, out = Path(temp) / "scan.pdf", Path(temp) / "out"
|
|
make_scan_pdf(pdf)
|
|
with patch.dict(os.environ, {"CONTENTPANEL_RENDER_SCALE": "1"}):
|
|
result = self.run_pipeline(pdf, out)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(json.loads(result.stdout)["ocr_pages"], [1])
|
|
self.assertEqual(len(_GatewayHandler.calls), 2)
|
|
self.assertEqual(_GatewayHandler.calls[1]["body"]["messages"][-1]["content"], text)
|
|
self.assertTrue((out / "2099-01-01-test_page_1.png").is_file())
|
|
|
|
def test_template_preserves_parts_escapes_text_and_assigns_unique_ids(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
target = Path(temp) / "filled.docx"
|
|
text = 'A & B <Tag> "Zitat" — Grüße'
|
|
fill_template(TEMPLATE, target, text, "02.09.2026", [text] * 60)
|
|
with zipfile.ZipFile(TEMPLATE) as source, zipfile.ZipFile(target) as result:
|
|
for name in source.namelist():
|
|
if name != "word/document.xml":
|
|
self.assertEqual(source.read(name), result.read(name), name)
|
|
document = ET.fromstring(result.read("word/document.xml"))
|
|
ns = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
|
|
texts = [node.text for node in document.findall(".//w:t", ns)]
|
|
self.assertEqual(texts.count(text), 61)
|
|
ids = [p.get("{http://schemas.microsoft.com/office/word/2010/wordml}paraId")
|
|
for p in document.findall(".//w:p", ns)]
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
with self.assertRaises(TemplateError):
|
|
fill_template(target, target.with_name("invalid.docx"), "Titel", "Datum", [text])
|
|
|
|
def test_ollama_native_structured_output_and_auth(self):
|
|
with Gateway(OLLAMA_MODEL="qwen3"):
|
|
result = transform("\n\n".join(_GatewayHandler.response_content["paragraphs"]))
|
|
self.assertEqual(result.title, "Gemeindefest am Sonntag")
|
|
self.assertEqual(len(result.paragraphs), 2)
|
|
call = _GatewayHandler.calls[0]
|
|
self.assertEqual(call["path"], "/api/chat")
|
|
self.assertEqual(call["authorization"], "Bearer test-key")
|
|
self.assertEqual(call["body"]["model"], "qwen3")
|
|
self.assertIsInstance(call["body"]["format"], dict)
|
|
self.assertIs(call["body"]["think"], False)
|
|
self.assertEqual(call['body']['options']['num_predict'], 16384)
|
|
|
|
def test_ollama_openai_compatible_mode(self):
|
|
with Gateway(style="openai"):
|
|
transform("\n\n".join(_GatewayHandler.response_content["paragraphs"]))
|
|
call = _GatewayHandler.calls[0]
|
|
self.assertEqual(call["path"], "/v1/chat/completions")
|
|
body = call["body"]
|
|
self.assertEqual(body["model"], "qwen3.8:27b")
|
|
self.assertEqual(body["response_format"]["type"], "json_schema")
|
|
self.assertIs(body["response_format"]["json_schema"]["strict"], True)
|
|
self.assertEqual(body["reasoning_effort"], "none")
|
|
self.assertEqual(body["max_tokens"], 16384)
|
|
self.assertIs(body["stream"], False)
|
|
|
|
def test_ollama_preserves_original_voice_lists_and_signoff(self):
|
|
paragraphs = ["Liebes Team,", "wir freuen uns auf Euch!", "1. Bringt gute Laune mit.",
|
|
"2. Kaffee steht bereit.", "Von Herzen\nEure Gastgeber"]
|
|
source = "\n\n".join(paragraphs)
|
|
with Gateway(response={"title": "Einladung", "paragraphs": paragraphs}, style="openai"):
|
|
self.assertEqual(transform(source).paragraphs, tuple(paragraphs))
|
|
sent = _GatewayHandler.calls[0]["body"]["messages"][-1]["content"]
|
|
self.assertTrue(sent.endswith(source))
|
|
self.assertNotIn("[Seite", sent)
|
|
|
|
def test_editorial_changes_are_kept_instead_of_restoring_the_source(self):
|
|
for source, paragraphs in (
|
|
("Einladung\n\nGemeindefest • Sonntag • Beginn 11:30 Uhr • Kaffee kostenlos",
|
|
["Wir laden Euch am Sonntag zum Gemeindefest ein. Beginn ist um 11:30 Uhr. Kaffee gibt es kostenlos."]),
|
|
("Kommt um 11:30 Uhr nach Berlin. Bringt bitte eine Tasse mit. Anmeldung nicht erforderlich.",
|
|
["• Beginn: 11:30 Uhr in Berlin\n• Bitte eine Tasse mitbringen\n• Keine Anmeldung erforderlich"]),
|
|
("Musik\n\nTermin am Sonntag. Kontakt: a@example.org\n\nMusik\n\nKontakt: a@example.org",
|
|
["Musik am Sonntag. Bei Fragen meldet Euch bei a@example.org."]),
|
|
("Herzlich wilkommen!\n\nSeite 1 von 2\n\nWir freuen uns sehr. Wir freuen uns sehr.",
|
|
["Herzlich willkommen! Wir freuen uns auf Euch."]),
|
|
("Lieber Freund, wir freuen uns sehr auf Deinen Besuch. Von Herzen Deine Freunde",
|
|
["Lieber Freund,", "wir freuen uns auf Deinen Besuch!", "Von Herzen\nDeine Freunde"]),
|
|
):
|
|
with self.subTest(source=source), Gateway(response={"title": "Einladung", "paragraphs": paragraphs}):
|
|
self.assertEqual(transform(source).paragraphs, tuple(paragraphs))
|
|
self.assertEqual(len(_GatewayHandler.calls), 1)
|
|
|
|
def test_truncated_gateway_responses_are_rejected(self):
|
|
with patch.object(_GatewayHandler, "finish_reason", "length"):
|
|
for style in ("native", "openai"):
|
|
with self.subTest(style=style), Gateway(style=style):
|
|
with self.assertRaisesRegex(OllamaError, "Tokenlimit"):
|
|
transform("\n\n".join(_GatewayHandler.response_content["paragraphs"]))
|
|
|
|
def test_date_is_deterministic_and_independent_of_locale(self):
|
|
months = ["Januar", "Februar", "März", "April", "Mai", "Juni", "Juli", "August", "September", "Oktober", "November", "Dezember"]
|
|
with patch.dict(os.environ, {"LC_ALL": "C", "LANG": "en_US.UTF-8"}):
|
|
for number, month in enumerate(months, 1):
|
|
self.assertEqual(_date_label(f"2026-{number:02d}-02"), f"02. {month} 2026")
|
|
with self.assertRaises(ValueError):
|
|
_date_label("2026-02-30")
|
|
|
|
def test_template_contains_no_page_numbers_or_cached_page_count(self):
|
|
with zipfile.ZipFile(TEMPLATE) as archive:
|
|
for name in archive.namelist():
|
|
if name.endswith(".xml"):
|
|
xml = archive.read(name).decode()
|
|
self.assertNotRegex(xml, r"\b(?:NUMPAGES|SECTIONPAGES|PAGE)\b|w:pgNumType|<Pages>")
|
|
|
|
def test_new_numeric_values_and_contact_details_are_rejected(self):
|
|
for original, changed in (
|
|
("Es werden 11 Personen erwartet.", "Es werden 12 Personen erwartet."),
|
|
("Beginn ist um 11:30 Uhr.", "Beginn ist um 12:45 Uhr."),
|
|
("Kontakt: a@example.org", "Kontakt: b@example.org"),
|
|
("Beitrag: 1.000,50 Euro", "Beitrag: 1.000,60 Euro"),
|
|
("Details: https://example.org/fest", "Details: https://example.org/reise"),
|
|
("Anmeldung: kirchenmusik.example.org", "Anmeldung: anderer.example.org"),
|
|
):
|
|
with self.subTest(changed=changed), Gateway(response={"title": "Gemeindefest", "paragraphs": [changed]}):
|
|
with self.assertRaisesRegex(OllamaError, "nicht belegte"):
|
|
transform(original)
|
|
|
|
def test_dates_can_use_numeric_months_and_leading_zeroes(self):
|
|
source = 'Konzert am 5. September 2026. Anmeldung bis 2. September 2026.'
|
|
paragraphs = ['• Konzert: 05.09.2026\n• Anmeldung bis 02.09.2026']
|
|
with Gateway(response={'title': 'Konzert', 'paragraphs': paragraphs}):
|
|
self.assertEqual(transform(source).paragraphs, tuple(paragraphs))
|
|
|
|
def test_new_title_cannot_invent_dates_numbers_or_contact_details(self):
|
|
source = "Wir laden Euch herzlich ein."
|
|
for title in ("Einladung am 13.09.2026", "Einladung um 12:30", "Kontakt a@example.org", "Einladung https://example.org"):
|
|
with self.subTest(title=title), Gateway(response={"title": title, "paragraphs": [source]}):
|
|
with self.assertRaisesRegex(OllamaError, "nicht belegte"):
|
|
transform(source)
|
|
|
|
def run_pipeline(self, pdf, output, mode="template"):
|
|
return subprocess.run([sys.executable, str(ROOT / "process.py"),
|
|
"--pdf", str(pdf), "--template", str(TEMPLATE), "--output-dir", str(output),
|
|
"--base-name", "2099-01-01-test", "--date", "2026-09-01", "--mode", mode],
|
|
capture_output=True, text=True, timeout=120)
|
|
|
|
def test_end_to_end_pipeline_and_overflow(self):
|
|
paragraphs = [("Dies ist ein längerer Testtext für die automatische Seitenaufteilung. " * 9).strip()] * 6
|
|
with Gateway(response={"title": "Mehrseitiger Testaushang", "paragraphs": paragraphs}), tempfile.TemporaryDirectory() as temp:
|
|
pdf, out = Path(temp) / "input.pdf", Path(temp) / "out"
|
|
make_text_pdf(pdf, "\n\n".join(paragraphs))
|
|
with patch.dict(os.environ, {"CONTENTPANEL_RENDER_SCALE": "1", "CONTENTPANEL_MAX_RENDERED_PAGES": "10"}):
|
|
process = self.run_pipeline(pdf, out)
|
|
self.assertEqual(process.returncode, 0, process.stderr)
|
|
manifest = json.loads(process.stdout)
|
|
self.assertGreaterEqual(manifest["pages"], 2)
|
|
pages = list(out.glob("2099-01-01-test_page_*.png"))
|
|
self.assertEqual(len(pages), manifest["pages"])
|
|
self.assertTrue(all(page.stat().st_size > 1000 for page in pages))
|
|
with patch.dict(os.environ, {"CONTENTPANEL_RENDER_SCALE": "1", "CONTENTPANEL_MAX_RENDERED_PAGES": "1"}):
|
|
failed = self.run_pipeline(pdf, Path(temp) / "rejected")
|
|
self.assertNotEqual(failed.returncode, 0)
|
|
self.assertIn("maximal 1", failed.stderr)
|
|
self.assertEqual(list((Path(temp) / "rejected").glob("*.png")), [])
|
|
|
|
def test_original_pdf_runs_without_gateway_and_preserves_all_pages(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
pdf, out = Path(temp) / "original.pdf", Path(temp) / "out"
|
|
document = FPDF()
|
|
for text in ("Erste Seite", "Letzte Seite"):
|
|
document.add_page()
|
|
document.set_font("Helvetica", size=24)
|
|
document.cell(0, 20, text)
|
|
document.output(str(pdf))
|
|
with patch.dict(os.environ, {"OLLAMA_GATEWAY_URL": "http://127.0.0.1:1"}):
|
|
result = self.run_pipeline(pdf, out, "original")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(json.loads(result.stdout)["pages"], 2)
|
|
text = subprocess.run(["tesseract", str(out / "2099-01-01-test_page_2.png"), "stdout", "-l", "deu"],
|
|
capture_output=True, text=True, check=True, timeout=30).stdout
|
|
self.assertIn("Letzte Seite", text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|