contentpanel - Aushangworkflow um docx Vorlage erweitern
This commit is contained in:
@@ -0,0 +1,161 @@
|
||||
"""Context overflow must split inputs, never silently discard their remainder."""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import requests
|
||||
|
||||
import test_pipeline as fixtures
|
||||
import ollama_client as client
|
||||
from pdf_text import _ocr_page
|
||||
|
||||
|
||||
def echo_request(base, model, source, timeout, style, system, schema, image, max_tokens):
|
||||
return {'title': 'Einladung', 'paragraphs': [source]}
|
||||
|
||||
|
||||
class ContextTests(unittest.TestCase):
|
||||
setUp = fixtures.PipelineTests.setUp
|
||||
|
||||
def test_character_limit_splits_in_order_and_keeps_every_word(self):
|
||||
source = '\n\n'.join(f'Absatz {i}: ' + 'Einladung ' * 35 for i in range(9))
|
||||
with patch.dict(os.environ, {'OLLAMA_MAX_INPUT_CHARS': '1000'}), \
|
||||
patch.object(client, '_request', side_effect=echo_request) as request:
|
||||
result = client.transform(source)
|
||||
self.assertGreater(request.call_count, 1)
|
||||
self.assertEqual(' '.join(result.paragraphs).split(), source.split())
|
||||
self.assertTrue(all(len(call.args[2]) <= 1000 for call in request.call_args_list))
|
||||
self.assertEqual(result.title, 'Einladung')
|
||||
|
||||
def test_byte_budget_protects_unicode_even_below_character_limit(self):
|
||||
source = 'Überraschung 🎉 漢字 ' * 1800
|
||||
sent = []
|
||||
def answer(*args):
|
||||
text, system = args[2], args[5]
|
||||
self.assertLessEqual(len(text.encode()) + len(system.encode()) + args[-1] + 2048, 32768)
|
||||
sent.append(text)
|
||||
return {'title': 'Einladung', 'paragraphs': ['Ankündigung.']}
|
||||
with patch.object(client, '_request', side_effect=answer):
|
||||
client.transform(source)
|
||||
self.assertGreater(len(sent), 1)
|
||||
self.assertEqual(' '.join(sent[:-1]).split(), source.split())
|
||||
self.assertTrue(sent[-1].startswith('Einladung')) # Smaller results get one final editorial pass.
|
||||
|
||||
def test_only_the_oversized_chunk_is_retried(self):
|
||||
source = 'A' * 700 + '\n\n' + 'B' * 700 + '\n\n' + 'C' * 700 + '\n\n' + 'D' * 700
|
||||
def answer(*args):
|
||||
if len(args[2]) > 2000 or (args[2].startswith('C') and len(args[2]) > 1000):
|
||||
raise client.ContextLimitError('Tokenlimit')
|
||||
return echo_request(*args)
|
||||
with patch.object(client, '_request', side_effect=answer) as request:
|
||||
result = client.transform(source)
|
||||
self.assertEqual(' '.join(result.paragraphs).split(), source.split())
|
||||
self.assertEqual([call.args[2] for call in request.call_args_list[:-1]],
|
||||
[source, source[:1402], source[1404:], 'C' * 700, 'D' * 700])
|
||||
self.assertTrue(request.call_args_list[-1].args[2].startswith('Einladung'))
|
||||
|
||||
def test_permanent_errors_do_not_retry_or_return_partial_content(self):
|
||||
source = 'A' * 700 + '\n\n' + 'B' * 700
|
||||
for error in (client.OllamaError('authentication'), client.ContentMismatchError('Fakten')):
|
||||
with self.subTest(error=error), patch.object(client, '_request', side_effect=error) as request:
|
||||
with self.assertRaises(type(error)):
|
||||
client.transform(source)
|
||||
self.assertEqual(request.call_count, 1)
|
||||
with patch.dict(os.environ, {'OLLAMA_MAX_INPUT_CHARS': '1000'}), patch.object(client, '_request',
|
||||
side_effect=[{'title': 'Einladung', 'paragraphs': ['Erster Teil.']}, client.OllamaError('offline')]):
|
||||
with self.assertRaisesRegex(client.OllamaError, 'offline'):
|
||||
client.transform(source)
|
||||
|
||||
def test_context_retry_has_a_bound(self):
|
||||
with patch.object(client, '_request', side_effect=client.ContextLimitError('Tokenlimit')) as request:
|
||||
with self.assertRaisesRegex(client.ContextLimitError, 'Tokenlimit'):
|
||||
client.transform('Text ' * 800)
|
||||
self.assertLessEqual(request.call_count, 4)
|
||||
|
||||
def test_facts_moved_into_section_titles_are_not_discarded(self):
|
||||
source = 'Fest im Garten. ' * 45 + '\n\n' + 'Abschluss im Saal. ' * 40
|
||||
with patch.dict(os.environ, {'OLLAMA_MAX_INPUT_CHARS': '1000'}), patch.object(client, '_request',
|
||||
side_effect=[{'title': 'Fest im Garten', 'paragraphs': ['Herzlich willkommen.']},
|
||||
{'title': 'Abschluss im Saal', 'paragraphs': ['Wir freuen uns auf Euch.']},
|
||||
{'title': 'Fest im Garten', 'paragraphs': ['Abschluss im Saal']}]) as request:
|
||||
result = client.transform(source)
|
||||
self.assertEqual(result.title, 'Fest im Garten')
|
||||
self.assertIn('Abschluss im Saal', result.paragraphs)
|
||||
self.assertIn('Abschluss im Saal', request.call_args_list[-1].args[2])
|
||||
|
||||
def test_invalid_final_edit_is_not_published_as_successful_sections(self):
|
||||
with patch.dict(os.environ, {'OLLAMA_MAX_INPUT_CHARS': '1000'}), patch.object(client, '_request',
|
||||
side_effect=[{'title': 'Einladung', 'paragraphs': ['Beginn im Garten.']},
|
||||
{'title': 'Einladung', 'paragraphs': ['Abschluss im Saal.']},
|
||||
client.OllamaError('invalid final response')]):
|
||||
with self.assertRaisesRegex(client.OllamaError, 'invalid final response'):
|
||||
client.transform('A' * 700 + '\n\n' + 'B' * 700)
|
||||
|
||||
def test_only_size_related_http_errors_enable_splitting(self):
|
||||
for status, message, expected in (
|
||||
(400, "This model's maximum context length is 32768 tokens", client.ContextLimitError),
|
||||
(422, 'input is too long', client.ContextLimitError),
|
||||
(413, 'Request Entity Too Large', client.ContextLimitError),
|
||||
(401, 'Unauthorized', client.OllamaError),
|
||||
(429, 'Rate limit exceeded', client.OllamaError),
|
||||
(500, 'out of memory', client.OllamaError),
|
||||
(400, 'model does not support images', client.OllamaError),
|
||||
):
|
||||
response = requests.Response()
|
||||
response.status_code = status
|
||||
response._content = json.dumps({'error': {'message': message}}).encode()
|
||||
self.assertIs(type(client._gateway_error(response)), expected)
|
||||
|
||||
def test_invalid_context_fails_before_any_request(self):
|
||||
for value in ('oops', '4096', '999999'):
|
||||
with self.subTest(value=value), patch.dict(os.environ, {'OLLAMA_CONTEXT_LENGTH': value}), \
|
||||
patch.object(client, '_request') as request:
|
||||
with self.assertRaises(client.OllamaError):
|
||||
client.transform('Einladung zum Fest.')
|
||||
request.assert_not_called()
|
||||
|
||||
def test_image_limit_splits_with_overlap_and_keeps_bottom(self):
|
||||
from pdf_text import run
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
pdf = Path(temp) / 'page.pdf'
|
||||
fixtures.make_text_pdf(pdf)
|
||||
for rotated in (False, True):
|
||||
def metadata(args, **kwargs):
|
||||
info = run(args, **kwargs)
|
||||
# Landscape MediaBox rotated by 90° must use its full width as height.
|
||||
if args[0] == 'pdfinfo' and rotated:
|
||||
info = re.sub(r'(Page\s+1\s+MediaBox:).*', r'\1 0 0 842 595', info)
|
||||
info = re.sub(r'(Page\s+1\s+rot:)\s+0', r'\1 90', info)
|
||||
return info
|
||||
with self.subTest(rotated=rotated), patch('pdf_text.run', side_effect=metadata) as commands, \
|
||||
patch('pdf_text.transcribe_image', side_effect=[
|
||||
client.ContextLimitError('Tokenlimit'), 'Anfang\nÜbergang', 'Übergang\nEnde']):
|
||||
result = _ocr_page(pdf, 1)
|
||||
crops = [call.args[0] for call in commands.call_args_list if call.args[0][0] == 'pdftoppm']
|
||||
self.assertNotIn('-y', crops[0])
|
||||
top = int(crops[1][crops[1].index('-y') + 1])
|
||||
height = int(crops[1][crops[1].index('-H') + 1])
|
||||
bottom_top = int(crops[2][crops[2].index('-y') + 1])
|
||||
bottom_height = int(crops[2][crops[2].index('-H') + 1])
|
||||
self.assertEqual(top, 0)
|
||||
self.assertEqual(height - bottom_top, 96)
|
||||
self.assertGreater(bottom_top + bottom_height, 2300)
|
||||
self.assertEqual(result, 'Anfang\nÜbergang\n\nÜbergang\nEnde')
|
||||
|
||||
def test_image_retry_does_not_retry_network_errors(self):
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
pdf = Path(temp) / 'page.pdf'
|
||||
fixtures.make_text_pdf(pdf)
|
||||
for error, max_calls in ((client.OllamaError('offline'), 1), (client.ContextLimitError('Tokenlimit'), 4)):
|
||||
with self.subTest(error=error), patch('pdf_text.transcribe_image', side_effect=error) as vision:
|
||||
with self.assertRaises(type(error)):
|
||||
_ocr_page(pdf, 1)
|
||||
self.assertEqual(vision.call_count, max_calls)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,449 @@
|
||||
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_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)
|
||||
@@ -0,0 +1,223 @@
|
||||
"""Real PostgreSQL + PHP + renderer; private files and captured mail, no real recipients."""
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import signal
|
||||
import subprocess
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from fpdf import FPDF
|
||||
|
||||
from test_pipeline import Gateway, _GatewayHandler, make_text_pdf
|
||||
|
||||
|
||||
@unittest.skipUnless(os.getenv('CONTENTPANEL_TEST_QUEUE_DSN'), 'Requires an isolated PostgreSQL test database')
|
||||
class QueueTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.temp = tempfile.TemporaryDirectory(prefix='contentpanel_queue_test_')
|
||||
cls.addClassCleanup(cls.temp.cleanup)
|
||||
cls.root = Path(cls.temp.name)
|
||||
insertion = Path('intranet_repo/projects/contentpanel/insertion')
|
||||
shutil.copytree(Path('/var/www/html') / insertion, cls.root / insertion, ignore=shutil.ignore_patterns('news'))
|
||||
cls.news = cls.root / insertion / 'news/png'
|
||||
cls.pdf = cls.root / 'input.pdf'
|
||||
cls.driver = cls.root / 'driver.php'
|
||||
cls.driver.write_text('''<?php
|
||||
require __DIR__ . '/intranet_repo/projects/contentpanel/insertion/util/contentpanel_helpers.php';
|
||||
function send_intranet_mail($recipient, $subject, $message): bool {
|
||||
if (file_exists(__DIR__ . '/mail-throw')) throw new RuntimeException('SMTP unavailable');
|
||||
$sent = !file_exists(__DIR__ . '/mail-fail');
|
||||
file_put_contents(__DIR__ . '/mail.jsonl', json_encode([$recipient, $subject, $message, $sent]) . "\\n", FILE_APPEND);
|
||||
return $sent;
|
||||
}
|
||||
$pdo = new PDO(getenv('CONTENTPANEL_TEST_QUEUE_DSN'), 'postgres', 'queue-test-only',
|
||||
[PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC]);
|
||||
contentpanel_queue_schema($pdo);
|
||||
switch ($argv[1]) {
|
||||
case 'setup':
|
||||
$pdo->exec("CREATE TABLE IF NOT EXISTS admin_messages (id BIGSERIAL PRIMARY KEY,
|
||||
type TEXT, sender_name TEXT, sender_email TEXT, subject TEXT, message TEXT,
|
||||
status TEXT CHECK (status IN ('pending', 'done')), created_at TIMESTAMPTZ, processed_at TIMESTAMPTZ)");
|
||||
$pdo->exec('TRUNCATE contentpanel_jobs, admin_messages RESTART IDENTITY');
|
||||
break;
|
||||
case 'enqueue':
|
||||
echo json_encode(contentpanel_enqueue_news($pdo, ['mail' => 'owner@example.test', 'first_name' => 'Test'],
|
||||
__DIR__ . '/input.pdf', ['base' => '2030-12-31-' . $argv[2], 'title' => $argv[2],
|
||||
'date' => '2026-09-02', 'mode' => 'template']));
|
||||
break;
|
||||
case 'work': echo json_encode(contentpanel_queue_once($pdo)); break;
|
||||
case 'query': echo json_encode($pdo->query($argv[2])->fetchAll()); break;
|
||||
}
|
||||
''', encoding='utf-8')
|
||||
|
||||
def php(self, *args, timeout=60):
|
||||
result = subprocess.run(['php', str(self.driver), *args], capture_output=True, text=True, timeout=timeout)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
return json.loads(result.stdout) if result.stdout else None
|
||||
|
||||
def query(self, sql):
|
||||
return self.php('query', sql)
|
||||
|
||||
def setUp(self):
|
||||
self.php('setup')
|
||||
for name in ('mail.jsonl', 'mail-fail', 'mail-throw'):
|
||||
(self.root / name).unlink(missing_ok=True)
|
||||
if self.news.exists():
|
||||
shutil.rmtree(self.news)
|
||||
make_text_pdf(self.pdf)
|
||||
self.gateway = Gateway()
|
||||
self.gateway.__enter__()
|
||||
self.addCleanup(self.gateway.__exit__, None, None, None)
|
||||
|
||||
def messages(self):
|
||||
return [json.loads(line) for line in (self.root / 'mail.jsonl').read_text().splitlines()]
|
||||
|
||||
def test_success_survives_upload_file_removal_and_logs_and_mails(self):
|
||||
self.assertIn('Hintergrund', self.php('enqueue', 'Erfolg'))
|
||||
self.pdf.unlink() # HTTP temporary files disappear when the upload response ends.
|
||||
self.assertTrue(self.php('work'))
|
||||
rows = self.query('SELECT pdf IS NULL AS cleared, result, mailed_at IS NOT NULL AS mailed FROM contentpanel_jobs')
|
||||
self.assertTrue(rows[0]['cleared'])
|
||||
self.assertTrue(rows[0]['mailed'])
|
||||
self.assertTrue(json.loads(rows[0]['result'])['ok'])
|
||||
self.assertEqual(self.messages()[0][0], 'owner@example.test')
|
||||
self.assertIn('veröffentlicht', self.messages()[0][1])
|
||||
inbox = self.query('SELECT * FROM admin_messages')
|
||||
self.assertEqual(len(inbox), 1)
|
||||
self.assertEqual(inbox[0]['type'], 'system')
|
||||
self.assertIn('Mailserver übergeben', inbox[0]['message'])
|
||||
self.assertTrue((self.news / '2030-12-31-Erfolg_page_1.png').is_file())
|
||||
self.assertFalse(list(self.news.glob('.upload-*')))
|
||||
|
||||
def test_failure_also_logs_and_mails_without_publishing(self):
|
||||
self.pdf.write_bytes(b'%PDF-broken')
|
||||
self.php('enqueue', 'Fehler')
|
||||
self.assertTrue(self.php('work'))
|
||||
row = self.query('SELECT result, pdf IS NULL AS cleared FROM contentpanel_jobs')[0]
|
||||
self.assertFalse(json.loads(row['result'])['ok'])
|
||||
self.assertTrue(row['cleared'])
|
||||
self.assertIn('fehlgeschlagen', self.messages()[0][1])
|
||||
self.assertNotIn('/opt/', self.messages()[0][2])
|
||||
self.assertEqual(len(self.query('SELECT * FROM admin_messages')), 1)
|
||||
self.assertFalse(list(self.news.glob('*.png')))
|
||||
|
||||
def test_mail_retry_does_not_render_or_log_the_job_again(self):
|
||||
(self.root / 'mail-fail').touch()
|
||||
self.php('enqueue', 'MailRetry')
|
||||
self.php('work')
|
||||
row = self.query('SELECT mailed_at FROM contentpanel_jobs')[0]
|
||||
self.assertIsNone(row['mailed_at'])
|
||||
self.assertIn('Versand fehlgeschlagen', self.query('SELECT message FROM admin_messages')[0]['message'])
|
||||
calls = len(_GatewayHandler.calls)
|
||||
(self.root / 'mail-fail').unlink()
|
||||
self.query("UPDATE contentpanel_jobs SET retry_at = NOW() - INTERVAL '1 second' RETURNING id")
|
||||
self.assertFalse(self.php('work'))
|
||||
self.assertEqual(len(_GatewayHandler.calls), calls)
|
||||
self.assertEqual(len(self.query('SELECT * FROM admin_messages')), 1)
|
||||
self.assertEqual([mail[3] for mail in self.messages()], [False, True])
|
||||
self.assertTrue(self.query('SELECT mailed_at IS NOT NULL AS sent FROM contentpanel_jobs')[0]['sent'])
|
||||
|
||||
def test_conversion_failure_publishes_all_original_pages(self):
|
||||
document = FPDF()
|
||||
for text in ('Originalanfang', 'Originalende'):
|
||||
document.add_page()
|
||||
document.set_font('Helvetica', size=18)
|
||||
document.multi_cell(0, 10, text + '. Bitte bewahrt die urspruengliche Gestaltung dieses Aushangs vollstaendig.')
|
||||
document.output(str(self.pdf))
|
||||
self.php('enqueue', 'OriginalFallback')
|
||||
with patch.object(_GatewayHandler, 'response_content', {'title': 'Erfundener Termin 2099', 'paragraphs': ['Einladung.']}):
|
||||
self.assertTrue(self.php('work'))
|
||||
result = json.loads(self.query('SELECT result FROM contentpanel_jobs')[0]['result'])
|
||||
self.assertTrue(result['ok'])
|
||||
self.assertTrue(result['fallback'])
|
||||
pages = sorted(self.news.glob('*OriginalFallback_page_*.png'))
|
||||
self.assertEqual(len(pages), 2)
|
||||
text = subprocess.run(['tesseract', str(pages[-1]), 'stdout', '-l', 'deu'],
|
||||
capture_output=True, text=True, check=True).stdout
|
||||
self.assertIn('Originalende', text)
|
||||
self.assertIn('Original-PDF', self.messages()[0][2])
|
||||
self.assertIn('Original-PDF', self.query('SELECT message FROM admin_messages')[0]['message'])
|
||||
|
||||
def test_mail_exception_does_not_block_the_next_job(self):
|
||||
(self.root / 'mail-throw').touch()
|
||||
self.php('enqueue', 'MailAusfall')
|
||||
self.assertTrue(self.php('work'))
|
||||
(self.root / 'mail-throw').unlink()
|
||||
self.query("UPDATE contentpanel_jobs SET retry_at = NOW() - INTERVAL '1 second' RETURNING id")
|
||||
self.php('enqueue', 'Weiter')
|
||||
self.assertTrue(self.php('work'))
|
||||
self.assertEqual(len(self.messages()), 2)
|
||||
self.assertTrue(all(row['sent'] for row in self.query('SELECT mailed_at IS NOT NULL AS sent FROM contentpanel_jobs')))
|
||||
|
||||
def blocked_worker(self):
|
||||
entered, release = threading.Event(), threading.Event()
|
||||
original = _GatewayHandler.do_POST
|
||||
def delayed(handler):
|
||||
entered.set()
|
||||
release.wait(20)
|
||||
try:
|
||||
original(handler)
|
||||
except (BrokenPipeError, ConnectionResetError):
|
||||
pass
|
||||
context = patch.object(_GatewayHandler, 'do_POST', delayed)
|
||||
context.start()
|
||||
self.addCleanup(context.stop)
|
||||
self.addCleanup(release.set)
|
||||
process = subprocess.Popen(['php', str(self.driver), 'work'], stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE, text=True, start_new_session=True)
|
||||
def cleanup():
|
||||
if process.poll() is None:
|
||||
os.killpg(process.pid, signal.SIGKILL)
|
||||
process.communicate(timeout=5)
|
||||
self.addCleanup(cleanup)
|
||||
self.assertTrue(entered.wait(10), 'Worker did not reach the gateway')
|
||||
return process, release
|
||||
|
||||
def test_global_lock_keeps_workers_serial_while_uploads_continue(self):
|
||||
self.php('enqueue', 'Zuerst')
|
||||
process, release = self.blocked_worker()
|
||||
start = time.monotonic()
|
||||
self.assertFalse(self.php('work', timeout=3))
|
||||
self.php('enqueue', 'Danach', timeout=3)
|
||||
self.assertLess(time.monotonic() - start, 3)
|
||||
self.assertEqual(self.query('SELECT attempts FROM contentpanel_jobs ORDER BY id'), [{'attempts': 1}, {'attempts': 0}])
|
||||
release.set()
|
||||
stdout, stderr = process.communicate(timeout=30)
|
||||
self.assertEqual(process.returncode, 0, stderr)
|
||||
self.assertEqual(stdout, 'true')
|
||||
self.assertTrue(self.php('work'))
|
||||
subjects = [row['subject'] for row in self.query('SELECT subject FROM admin_messages ORDER BY id')]
|
||||
self.assertTrue(subjects[0].endswith('Zuerst'))
|
||||
self.assertTrue(subjects[1].endswith('Danach'))
|
||||
|
||||
def test_killed_worker_is_recovered_without_duplicate_completion(self):
|
||||
self.php('enqueue', 'Neustart')
|
||||
process, release = self.blocked_worker()
|
||||
os.killpg(process.pid, signal.SIGKILL)
|
||||
process.communicate(timeout=5)
|
||||
release.set()
|
||||
self.assertIsNone(self.query('SELECT result FROM contentpanel_jobs')[0]['result'])
|
||||
self.assertTrue(self.php('work'))
|
||||
self.assertEqual(self.query('SELECT attempts FROM contentpanel_jobs')[0]['attempts'], 2)
|
||||
self.assertEqual(len(self.query('SELECT * FROM admin_messages')), 1)
|
||||
self.assertEqual(len(self.messages()), 1)
|
||||
self.assertFalse(list(self.news.glob('.upload-*')))
|
||||
|
||||
def test_repeated_interruptions_use_original_and_expired_jobs_fail_without_llm(self):
|
||||
for change in ("attempts = 2", "payload = jsonb_set(payload, '{base}', '\"2000-01-01-Abgelaufen\"')"):
|
||||
self.php('enqueue', 'Abbruch')
|
||||
self.query('UPDATE contentpanel_jobs SET ' + change + ' WHERE result IS NULL RETURNING id')
|
||||
self.assertTrue(self.php('work'))
|
||||
self.assertEqual(_GatewayHandler.calls, [])
|
||||
self.assertEqual(len(self.messages()), 2)
|
||||
self.assertIn('Original-PDF', self.messages()[0][2])
|
||||
self.assertIn('fehlgeschlagen', self.messages()[1][1])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main(verbosity=2)
|
||||
@@ -0,0 +1,170 @@
|
||||
"""Integration checks against the production Chromium/Rust renderer, no LLM."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from playwright.sync_api import sync_playwright
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT))
|
||||
from betteroffice_render import RenderError, _check_renderer, _server, render
|
||||
from docx_template import fill_template
|
||||
from process import _date_label
|
||||
|
||||
TEMPLATE = Path(os.getenv("CONTENTPANEL_TEST_TEMPLATE", "/var/www/html/intranet_repo/projects/contentpanel/insertion/template/aushang-template.docx"))
|
||||
|
||||
|
||||
class RenderingTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.temp = tempfile.TemporaryDirectory(prefix="contentpanel_render_test_")
|
||||
cls.output = Path(os.getenv("CONTENTPANEL_TEST_OUTPUT", cls.temp.name))
|
||||
cls.output.mkdir(parents=True, exist_ok=True)
|
||||
cls.document = cls.output / "verified.docx"
|
||||
with zipfile.ZipFile(TEMPLATE) as archive:
|
||||
(cls.output / "original-logo.png").write_bytes(archive.read("word/media/image1.png"))
|
||||
paragraphs = [
|
||||
("Dies ist ein langer Testabsatz zur Prüfung von Schriftgröße, Zeilenumbruch und Seitenaufteilung. " * 12).strip()
|
||||
for _ in range(3)
|
||||
] + ["Erste Gedichtzeile\nZweite Gedichtzeile\n\nVon Herzen\nEure Gastgeber",
|
||||
"Abschlussmarker: Dieser letzte Absatz muss vollständig auf der letzten Seite erscheinen."]
|
||||
fill_template(TEMPLATE, cls.document,
|
||||
"Verabschiedung von Diakon Georg aus der Gemeinde Reinickendorf",
|
||||
_date_label("2026-09-02"), paragraphs)
|
||||
with patch.dict(os.environ, {"CONTENTPANEL_RENDER_SCALE": "1"}):
|
||||
cls.page_count = render(cls.document, cls.output)
|
||||
cls.metrics = []
|
||||
server = _server(cls.output, cls.document, ROOT / "fonts")
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
with sync_playwright() as playwright:
|
||||
browser = playwright.chromium.launch(headless=True, args=["--disable-dev-shm-usage"])
|
||||
page = browser.new_page()
|
||||
for number in range(1, cls.page_count + 1):
|
||||
page.goto(f"http://127.0.0.1:{server.server_address[1]}/page-{number}.png")
|
||||
page.wait_for_function("document.querySelector('img')?.complete")
|
||||
cls.metrics.append(page.evaluate("""async () => {
|
||||
const img = document.querySelector('img');
|
||||
const canvas = document.createElement('canvas');
|
||||
canvas.width = img.naturalWidth; canvas.height = img.naturalHeight;
|
||||
const ctx = canvas.getContext('2d'); ctx.drawImage(img, 0, 0);
|
||||
const {data} = ctx.getImageData(0, 0, canvas.width, canvas.height);
|
||||
const blue = {left:canvas.width, top:canvas.height, right:0, bottom:0, count:0};
|
||||
const rows = Array(canvas.height).fill(0);
|
||||
let rightOverflow = 0;
|
||||
for (let y=0;y<canvas.height;y++) for (let x=0;x<canvas.width;x++) {
|
||||
const i=(y*canvas.width+x)*4, r=data[i], g=data[i+1], b=data[i+2];
|
||||
if (b>130 && r<90 && g>60 && b>g+20) {
|
||||
blue.left=Math.min(blue.left,x); blue.right=Math.max(blue.right,x);
|
||||
blue.top=Math.min(blue.top,y); blue.bottom=Math.max(blue.bottom,y); blue.count++;
|
||||
}
|
||||
if (r<100 && g<100 && b<100) {
|
||||
if (x>=50 && x<725) rows[y]++;
|
||||
if (x>=725 && y>300 && y<930) rightOverflow++;
|
||||
}
|
||||
}
|
||||
const bands=[]; let start=null;
|
||||
rows.forEach((count,y)=>{
|
||||
if(count>=4 && start===null) start=y;
|
||||
if((count<4 || y===rows.length-1) && start!==null) {
|
||||
bands.push({top:start,bottom:y-1,height:y-start}); start=null;
|
||||
}
|
||||
});
|
||||
// Compare the full white symbol, not just its blue box.
|
||||
// A shifted duplicate used to pass the position check.
|
||||
const reference = new Image(); reference.src = '/original-logo.png';
|
||||
await reference.decode();
|
||||
const refCanvas = document.createElement('canvas');
|
||||
const w=blue.right-blue.left+1, h=blue.bottom-blue.top+1;
|
||||
refCanvas.width=w; refCanvas.height=h;
|
||||
const refCtx=refCanvas.getContext('2d'); refCtx.drawImage(reference,0,0,w,h);
|
||||
const ref=refCtx.getImageData(0,0,w,h).data;
|
||||
const actual=ctx.getImageData(blue.left,blue.top,w,h).data;
|
||||
const white=(pixels,i)=>pixels[i]>150 && pixels[i+1]>180 && pixels[i+2]>200;
|
||||
let tp=0, fp=0, fn=0;
|
||||
for(let i=0;i<ref.length;i+=4) {
|
||||
const expected=white(ref,i), rendered=white(actual,i);
|
||||
if(expected && rendered) tp++;
|
||||
else if(rendered) fp++;
|
||||
else if(expected) fn++;
|
||||
}
|
||||
const logoF1=2*tp/(2*tp+fp+fn);
|
||||
return {width:canvas.width,height:canvas.height,blue,bands,rightOverflow,logoF1};
|
||||
}"""))
|
||||
browser.close()
|
||||
finally:
|
||||
server.shutdown(); server.server_close(); thread.join(timeout=2)
|
||||
(cls.output / "pixel-metrics.json").write_text(json.dumps(cls.metrics, indent=2), encoding="utf-8")
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.temp.cleanup()
|
||||
|
||||
def test_long_text_paginates_and_last_paragraph_survives(self):
|
||||
import subprocess
|
||||
self.assertGreaterEqual(self.page_count, 2)
|
||||
self.assertLessEqual(self.page_count, 4)
|
||||
text = subprocess.run(["tesseract", str(self.output / f"page-{self.page_count}.png"),
|
||||
"stdout", "-l", "deu", "--psm", "11"],
|
||||
capture_output=True, text=True, check=True, timeout=30).stdout
|
||||
self.assertIn("Abschlussmarker", text)
|
||||
|
||||
def test_real_pixels_have_large_title_and_wrapped_body(self):
|
||||
first = self.metrics[0]
|
||||
title = [band for band in first["bands"] if 90 < band["top"] < 275 and band["height"] > 10]
|
||||
self.assertEqual(len(title), 3)
|
||||
self.assertTrue(all(band["height"] >= 30 for band in title), title)
|
||||
body = [band for band in first["bands"] if 325 < band["top"] < 930]
|
||||
self.assertGreater(len(body), 15)
|
||||
self.assertTrue(all(metric["rightOverflow"] == 0 for metric in self.metrics))
|
||||
|
||||
def test_logo_is_on_the_right_and_complete_on_every_page(self):
|
||||
for number, metric in enumerate(self.metrics, 1):
|
||||
with self.subTest(page=number):
|
||||
self.assertEqual((metric["width"], metric["height"]), (794, 1123))
|
||||
logo = metric["blue"]
|
||||
self.assertGreater(logo["count"], 3500)
|
||||
self.assertAlmostEqual(logo["left"], 643, delta=2)
|
||||
self.assertAlmostEqual(logo["right"], 718, delta=2)
|
||||
self.assertGreater(logo["top"], 950)
|
||||
self.assertLess(logo["bottom"], 1060)
|
||||
self.assertAlmostEqual(logo["right"]-logo["left"], logo["bottom"]-logo["top"], delta=2)
|
||||
self.assertGreater(metric["logoF1"], .78, metric)
|
||||
|
||||
def test_missing_font_stops_before_png_export(self):
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
with patch.dict(os.environ, {"CONTENTPANEL_FONT_DIR": temp}):
|
||||
with self.assertRaisesRegex(RenderError, "Schriftdatei fehlt"):
|
||||
render(self.document, Path(temp) / "output")
|
||||
self.assertFalse(list(Path(temp).rglob("*.png")))
|
||||
|
||||
def test_corrupt_font_stops_before_png_export(self):
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
fonts = Path(temp) / "fonts"
|
||||
shutil.copytree(ROOT / "fonts", fonts)
|
||||
(fonts / "SuisseIntl-Light.otf").write_bytes(b"invalid font")
|
||||
with patch.dict(os.environ, {"CONTENTPANEL_FONT_DIR": str(fonts)}):
|
||||
with self.assertRaises(RenderError):
|
||||
render(self.document, Path(temp) / "output")
|
||||
self.assertFalse(list(Path(temp).rglob("*.png")))
|
||||
|
||||
def test_synthetic_metrics_are_not_accepted_as_success(self):
|
||||
class Page:
|
||||
def evaluate(self, _):
|
||||
return None
|
||||
with self.assertRaisesRegex(RenderError, "synthetic text metrics"):
|
||||
_check_renderer(Page(), ["warning: [canvasBackend] synthetic text metrics active"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
@@ -0,0 +1,332 @@
|
||||
"""Exercise the real PHP page in an isolated document root; never touch live data."""
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
import urllib.request
|
||||
import zipfile
|
||||
from unittest.mock import patch
|
||||
|
||||
from playwright.sync_api import sync_playwright
|
||||
from test_pipeline import Gateway, _GatewayHandler, make_text_pdf
|
||||
|
||||
WEB = Path('/var/www/html')
|
||||
INSERTION = Path('intranet_repo/projects/contentpanel/insertion')
|
||||
CENTRAL = Path('intranet_repo/projects/_central')
|
||||
|
||||
|
||||
class WebPreviewTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.temp = tempfile.TemporaryDirectory(prefix='contentpanel_web_test_')
|
||||
cls.addClassCleanup(cls.temp.cleanup)
|
||||
cls.root = Path(cls.temp.name)
|
||||
shutil.copytree(WEB / INSERTION, cls.root / INSERTION, ignore=shutil.ignore_patterns('news'))
|
||||
shutil.copytree(WEB / CENTRAL / 'layouts', cls.root / CENTRAL / 'layouts')
|
||||
shutil.copytree(WEB / CENTRAL / 'fonts', cls.root / CENTRAL / 'fonts')
|
||||
(cls.root / CENTRAL / 'scripts').mkdir(parents=True)
|
||||
shutil.copyfile(WEB / CENTRAL / 'scripts/contentpanel.js', cls.root / CENTRAL / 'scripts/contentpanel.js')
|
||||
bootstrap = cls.root / 'intranet_repo/general_files/initialise.php'
|
||||
bootstrap.parent.mkdir(parents=True)
|
||||
shutil.copyfile(Path(__file__).with_name('web_fixture.php'), bootstrap)
|
||||
(cls.root / CENTRAL / 'php').mkdir()
|
||||
for name in ('html_head', 'header', 'footer'):
|
||||
(cls.root / CENTRAL / f'php/{name}.php').write_text(
|
||||
'<meta charset="utf-8"><link rel="stylesheet" href="/intranet_repo/projects/_central/layouts/nak-design.css">'
|
||||
if name == 'html_head' else '', encoding='utf-8')
|
||||
cls.news = cls.root / INSERTION / 'news/png'
|
||||
cls.news.mkdir(parents=True)
|
||||
cls.helper = cls.root / INSERTION / 'util/contentpanel_helpers.php'
|
||||
with zipfile.ZipFile(WEB / INSERTION / 'template/aushang-template.docx') as archive:
|
||||
cls.image = archive.read('word/media/image1.png')
|
||||
cls.publish('2030-12-31-Mehrseitig', 10)
|
||||
cls.publish('2030-12-31-Einzelseite', 1)
|
||||
(cls.news / '2030-12-31-Mehrseitig_page_invalid.png').write_bytes(cls.image)
|
||||
cls.pdf = cls.root / 'input.pdf'
|
||||
make_text_pdf(cls.pdf)
|
||||
with socket.socket() as sock:
|
||||
sock.bind(('127.0.0.1', 0))
|
||||
port = sock.getsockname()[1]
|
||||
cls.url = f'http://127.0.0.1:{port}/{INSERTION}/insert.php'
|
||||
cls.gateway = Gateway(response={'title': 'Gemeindefest am Sonntag', 'paragraphs': [
|
||||
'Beginn: 11:30 Uhr', 'Ort: Berlin-Reinickendorf', 'Herzliche Einladung nach dem Gottesdienst.']})
|
||||
cls.gateway.__enter__()
|
||||
cls.addClassCleanup(cls.gateway.__exit__, None, None, None)
|
||||
cls.server = subprocess.Popen(['php', '-S', f'127.0.0.1:{port}', '-t', str(cls.root)],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
cls.addClassCleanup(cls.stop_server)
|
||||
for _ in range(50):
|
||||
try:
|
||||
urllib.request.urlopen(cls.url, timeout=1).close()
|
||||
break
|
||||
except OSError:
|
||||
time.sleep(.1)
|
||||
else:
|
||||
raise RuntimeError('PHP test server did not start')
|
||||
cls.playwright = sync_playwright().start()
|
||||
cls.addClassCleanup(cls.playwright.stop)
|
||||
cls.browser = cls.playwright.chromium.launch(headless=True, args=['--disable-dev-shm-usage'])
|
||||
cls.addClassCleanup(cls.browser.close)
|
||||
|
||||
@classmethod
|
||||
def stop_server(cls):
|
||||
cls.server.terminate()
|
||||
cls.server.wait(timeout=5)
|
||||
|
||||
@classmethod
|
||||
def php(cls, code, *args, check=True):
|
||||
return subprocess.run(['php', '-r', 'require $argv[1]; ' + code, str(cls.helper), *args],
|
||||
check=check, capture_output=True, text=True, timeout=10)
|
||||
|
||||
@classmethod
|
||||
def publish(cls, base, count):
|
||||
source = cls.root / 'generated'
|
||||
source.mkdir(exist_ok=True)
|
||||
files = [source / f'{base}_page_{number}.png' for number in range(1, count + 1)]
|
||||
for path in files:
|
||||
path.write_bytes(cls.image)
|
||||
cls.php('contentpanel_publish_news_pages(json_decode($argv[2], true), $argv[3]);',
|
||||
json.dumps([str(path) for path in files]), base)
|
||||
|
||||
def setUp(self):
|
||||
self.page = self.browser.new_page(viewport={'width': 1280, 'height': 1000})
|
||||
self.addCleanup(self.page.close)
|
||||
self.errors = []
|
||||
self.page.on('pageerror', lambda error: self.errors.append(str(error)))
|
||||
self.page.goto(self.url + '#dateien')
|
||||
|
||||
def tearDown(self):
|
||||
self.assertEqual(self.errors, [])
|
||||
|
||||
def test_page_has_simple_preview_and_working_tabs_on_desktop_and_mobile(self):
|
||||
for width in (1280, 390):
|
||||
self.page.set_viewport_size({'width': width, 'height': 1000})
|
||||
card = self.page.locator('.sb-aushang-card', has_text='Mehrseitig')
|
||||
self.assertEqual(card.locator('img').count(), 1)
|
||||
self.assertIn('_page_1.png?', card.locator('img').get_attribute('src'))
|
||||
self.assertIn('10 Seiten', card.inner_text())
|
||||
self.assertIn('1 Seite', self.page.locator('.sb-aushang-card', has_text='Einzelseite').inner_text())
|
||||
self.assertEqual(card.locator('a, details, [data-poster-navigation]').count(), 0)
|
||||
self.assertNotIn('Alle Seiten öffnen', self.page.content())
|
||||
self.page.wait_for_function('[...document.querySelectorAll(".sb-aushang-card img")].every(i => i.complete && i.naturalWidth)')
|
||||
for tab in ('gottesdienst', 'ansagen', 'dateien'):
|
||||
self.page.locator(f'[data-tab="{tab}"]').click()
|
||||
self.assertTrue(self.page.locator(f'#tab-{tab}').is_visible())
|
||||
self.assertTrue(self.page.evaluate('document.documentElement.scrollWidth <= innerWidth'),
|
||||
self.page.evaluate('[...document.querySelectorAll("body *")].filter(e => e.getBoundingClientRect().right > innerWidth).map(e => e.className)'))
|
||||
output = os.getenv('CONTENTPANEL_TEST_OUTPUT')
|
||||
if output:
|
||||
self.page.screenshot(path=str(Path(output) / f'web-{width}.png'), full_page=True, animations='disabled')
|
||||
self.assertEqual(self.page.locator('script:not([src])').count(), 0)
|
||||
|
||||
def test_permissions_and_csrf_reject_mutations(self):
|
||||
denied = self.page.request.post(self.url + '?deny=1', form={'action': 'delete_news', 'file_base': '2030-12-31-Mehrseitig'}, max_redirects=0)
|
||||
self.assertEqual(denied.status, 303)
|
||||
self.assertIn('/intranet/index.php', denied.headers['location'])
|
||||
self.page.request.post(self.url, form={'action': 'delete_news', 'file_base': '2030-12-31-Mehrseitig', 'csrf_token': 'wrong'}, max_redirects=0)
|
||||
self.page.reload()
|
||||
self.assertIn('Ungültige Anfrage', self.page.locator('.sb-alert-error').inner_text())
|
||||
self.assertTrue(all((self.news / f'2030-12-31-Mehrseitig_page_{n}.png').is_file() for n in range(1, 11)))
|
||||
|
||||
def test_forms_preserve_fields_validation_and_bound_sql_parameters(self):
|
||||
for tab, action, fields in (
|
||||
('gottesdienst', 'create_service', {'bibeldatum': '2030-12-31', 'bibelwort': 'Joh 3,16', 'bibelvers': 'Text & "Zitat"'}),
|
||||
('ansagen', 'create_announcement', {'ansage_titel': 'Gemeindefest', 'ansage_datum_von': '2030-12-31', 'ansage_ort': 'Saal'}),
|
||||
):
|
||||
self.page.locator(f'[data-tab="{tab}"]').click()
|
||||
form = self.page.locator('form').filter(has=self.page.locator(f'input[name="action"][value="{action}"]'))
|
||||
for name, value in fields.items():
|
||||
form.locator(f'[name="{name}"]').fill(value)
|
||||
form.locator('button[type="submit"]').click()
|
||||
self.page.wait_for_selector(f'#tab-{tab}.active .sb-alert')
|
||||
self.assertIn('erfolgreich angelegt', self.page.locator('.sb-alert').inner_text())
|
||||
writes = [json.loads(line) for line in (self.root / 'writes.jsonl').read_text().splitlines()]
|
||||
self.assertEqual(writes[-2][1], ['Joh 3,16', 'Text & "Zitat"', '2030-12-31'])
|
||||
self.assertEqual(writes[-1][1], ['2030-12-31', None, None, 'Saal', 'Gemeindefest', ''])
|
||||
self.page.locator('#ansage_titel').fill('Ungültig')
|
||||
self.page.locator('#ansage_datum_von').fill('2030-12-31')
|
||||
self.page.locator('#ansage_datum_bis').fill('2030-01-01')
|
||||
self.page.locator('#tab-ansagen').get_by_role('button', name='Eintragen').click()
|
||||
self.page.wait_for_selector('.sb-alert-error')
|
||||
self.assertIn('Enddatum darf nicht', self.page.locator('.sb-alert-error').inner_text())
|
||||
|
||||
def test_original_upload_and_delete_through_real_page(self):
|
||||
self.page.locator('#news_date').fill('2030-12-31')
|
||||
self.page.locator('#news_name').fill('Uploadtest')
|
||||
self.page.locator('#fileToUpload').set_input_files(self.pdf)
|
||||
self.page.get_by_role('button', name='Datei hochladen').click()
|
||||
self.page.wait_for_selector('.sb-alert', timeout=60000)
|
||||
self.assertEqual('Aushang wurde veröffentlicht.', self.page.locator('.sb-alert').inner_text())
|
||||
card = self.page.locator('.sb-aushang-card', has_text='Uploadtest')
|
||||
self.assertEqual(card.count(), 1)
|
||||
self.page.once('dialog', lambda dialog: dialog.dismiss())
|
||||
card.get_by_role('button', name='Löschen').click()
|
||||
self.assertEqual(card.count(), 1)
|
||||
self.page.once('dialog', lambda dialog: dialog.accept())
|
||||
card.get_by_role('button', name='Löschen').click()
|
||||
self.page.wait_for_selector('.sb-alert')
|
||||
self.assertEqual(card.count(), 0)
|
||||
self.assertFalse(list(self.news.glob('*Uploadtest*')))
|
||||
|
||||
def prepare_template_upload(self):
|
||||
self.page.locator('#news_date').fill('2030-12-31')
|
||||
self.page.locator('#news_name').fill('Fortschritt')
|
||||
self.page.locator('[name="format_mode"][value="template"]').check()
|
||||
self.page.locator('#fileToUpload').set_input_files(self.pdf)
|
||||
|
||||
def test_template_upload_queues_immediately_and_allows_another_upload(self):
|
||||
calls = len(_GatewayHandler.calls)
|
||||
states = []
|
||||
self.page.on('console', lambda message: states.append(json.loads(message.text[10:]))
|
||||
if message.text.startswith('upload-ui:') else None)
|
||||
for title in ('Erster-Auftrag', 'Zweiter-Auftrag'):
|
||||
self.prepare_template_upload()
|
||||
self.page.locator('#news_name').fill(title)
|
||||
self.page.evaluate('''() => {
|
||||
const field = document.createElement('input');
|
||||
field.name = 'mail'; field.value = 'forged@example.test'; field.type = 'hidden';
|
||||
document.querySelector('#tab-dateien [data-upload-form]').append(field);
|
||||
}''')
|
||||
self.page.evaluate('''() => document.querySelector('#tab-dateien [data-upload-form]').addEventListener('submit', () => {
|
||||
const form = document.querySelector('#tab-dateien [data-upload-form]');
|
||||
console.log('upload-ui:' + JSON.stringify([
|
||||
getComputedStyle(form.querySelector('.sb-spinner')).display,
|
||||
form.querySelector('[type=submit]').disabled]));
|
||||
})''')
|
||||
self.page.get_by_role('button', name='Datei hochladen').click()
|
||||
self.page.wait_for_selector('.sb-alert')
|
||||
self.assertEqual(states[-1], ['block', True])
|
||||
self.assertIn('erfolgreich hochgeladen', self.page.locator('.sb-alert').inner_text())
|
||||
self.assertIn('Hintergrund', self.page.locator('.sb-alert').inner_text())
|
||||
self.assertIn('E-Mail', self.page.locator('.sb-alert').inner_text())
|
||||
self.assertFalse(list(self.news.glob('*' + title + '*')))
|
||||
self.assertTrue(self.page.get_by_role('button', name='Datei hochladen').is_enabled())
|
||||
self.assertEqual(self.page.locator('.sb-progress, .sb-processing-note').count(), 0)
|
||||
writes = [json.loads(line) for line in (self.root / 'writes.jsonl').read_text().splitlines()]
|
||||
sql, parameters = writes[-1]
|
||||
self.assertIn('INSERT INTO contentpanel_jobs', sql)
|
||||
self.assertEqual(json.loads(parameters[0])['mail'], 'member@example.test')
|
||||
import base64
|
||||
self.assertEqual(base64.b64decode(parameters[1]), self.pdf.read_bytes())
|
||||
self.assertEqual(len(_GatewayHandler.calls), calls)
|
||||
|
||||
def test_invalid_upload_is_not_queued_and_can_be_retried(self):
|
||||
self.prepare_template_upload()
|
||||
self.page.locator('#fileToUpload').set_input_files({'name': 'broken.pdf', 'mimeType': 'application/pdf', 'buffer': b'no pdf'})
|
||||
self.page.get_by_role('button', name='Datei hochladen').click()
|
||||
self.page.wait_for_selector('.sb-alert-error')
|
||||
self.assertIn('gültige PDF', self.page.locator('.sb-alert-error').inner_text())
|
||||
self.assertTrue(self.page.get_by_role('button', name='Datei hochladen').is_enabled())
|
||||
|
||||
def test_queued_upload_still_requires_permission_and_csrf(self):
|
||||
form = {'action': 'upload_news', 'format_mode': 'template', 'csrf_token': 'wrong'}
|
||||
denied = self.page.request.post(self.url + '?deny=1', form=form, max_redirects=0)
|
||||
self.assertEqual(denied.status, 303)
|
||||
response = self.page.request.post(self.url, form=form, max_redirects=0)
|
||||
self.assertEqual(response.status, 303)
|
||||
self.page.reload()
|
||||
self.assertIn('Ungültige Anfrage', self.page.locator('.sb-alert-error').inner_text())
|
||||
|
||||
def test_republishing_fewer_pages_and_failed_publish_preserve_existing_files(self):
|
||||
base = '2030-12-31-Austausch'
|
||||
self.publish(base, 10)
|
||||
self.publish(base, 2)
|
||||
result = self.php('echo json_encode(contentpanel_list_news());')
|
||||
item = next(item for item in json.loads(result.stdout) if item['base'] == base)
|
||||
self.assertEqual(item['pages'], 2)
|
||||
self.assertEqual(len(list(self.news.glob(base + '_page_*.png'))), 2)
|
||||
existing = {p.name: p.read_bytes() for p in self.news.glob(base + '_page_*.png')}
|
||||
failed = self.php('contentpanel_publish_news_pages([$argv[2]], $argv[3]);', '/missing/page.png', base, check=False)
|
||||
self.assertNotEqual(failed.returncode, 0)
|
||||
self.assertEqual(existing, {p.name: p.read_bytes() for p in self.news.glob(base + '_page_*.png')})
|
||||
self.assertFalse(list(self.news.glob('*.tmp-*')))
|
||||
self.assertEqual(self.php('echo json_encode(contentpanel_delete_news_files("../../other"));').stdout, 'false')
|
||||
|
||||
def test_publication_rolls_back_after_partial_rename(self):
|
||||
base = '2030-12-31-Rollback'
|
||||
self.publish(base, 2)
|
||||
original = {p.name: p.read_bytes() for p in self.news.glob(base + '_page_*.png')}
|
||||
source = self.root / 'rollback-input'
|
||||
source.mkdir()
|
||||
files = [source / f'{base}_page_{n}.png' for n in (1, 2)]
|
||||
for path in files:
|
||||
path.write_bytes(b'new image')
|
||||
# Fail exactly the second publish rename, after the first page was replaced.
|
||||
# Namespaced function injection leaves application code and real files untouched.
|
||||
injected = self.root / INSERTION / 'util/failing-news.php'
|
||||
prefix = r"""<?php namespace Fault; use \RuntimeException; use \Throwable;
|
||||
function rename($from, $to) {
|
||||
if (str_contains($from, '/rollback-input/') && str_ends_with($to, '_page_2.png')) return false;
|
||||
return \rename($from, $to);
|
||||
}
|
||||
"""
|
||||
injected.write_text(prefix + (self.helper.parent / 'news.php').read_text().removeprefix('<?php'), encoding='utf-8')
|
||||
result = self.php('require $argv[2]; Fault\\contentpanel_publish_news_pages(json_decode($argv[3], true), $argv[4]);',
|
||||
str(injected), json.dumps([str(path) for path in files]), base, check=False)
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn('Aushangseite konnte nicht veröffentlicht', result.stderr)
|
||||
self.assertEqual(original, {p.name: p.read_bytes() for p in self.news.glob(base + '_page_*.png')})
|
||||
self.assertFalse(list(self.news.glob(base + '*.bak-*')))
|
||||
|
||||
def test_delete_waits_for_publication_lock(self):
|
||||
import fcntl
|
||||
base = '2030-12-31-Sperre'
|
||||
self.publish(base, 1)
|
||||
with (self.news / '.publish.lock').open('w') as lock:
|
||||
fcntl.flock(lock, fcntl.LOCK_EX)
|
||||
process = subprocess.Popen(['php', '-r',
|
||||
'require $argv[1]; echo "ready\\n"; contentpanel_delete_news_files($argv[2]);', str(self.helper), base],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
||||
try:
|
||||
self.assertEqual(process.stdout.readline().strip(), 'ready')
|
||||
self.assertIsNone(process.poll())
|
||||
self.assertTrue((self.news / f'{base}_page_1.png').is_file())
|
||||
finally:
|
||||
fcntl.flock(lock, fcntl.LOCK_UN)
|
||||
process.communicate(timeout=5)
|
||||
self.assertEqual(process.returncode, 0)
|
||||
self.assertFalse((self.news / f'{base}_page_1.png').exists())
|
||||
|
||||
def test_delete_actions_bind_ids_and_reject_invalid_values(self):
|
||||
for tab, action, table in (('gottesdienst', 'delete_service', 'gottesdienst'), ('ansagen', 'delete_announcement', 'ansagen')):
|
||||
self.page.locator(f'[data-tab="{tab}"]').click()
|
||||
form = self.page.locator('form').filter(has=self.page.locator(f'[name="action"][value="{action}"]'))
|
||||
self.page.once('dialog', lambda dialog: dialog.accept())
|
||||
form.get_by_role('button', name='Löschen').click()
|
||||
self.page.wait_for_selector(f'#tab-{tab} .sb-alert')
|
||||
sql, params = json.loads((self.root / 'writes.jsonl').read_text().splitlines()[-1])
|
||||
self.assertEqual(sql, f'DELETE FROM {table} WHERE id = ?')
|
||||
self.assertEqual(len(params), 1)
|
||||
self.assertIsInstance(params[0], int)
|
||||
failed = self.php('contentpanel_positive_id("1 OR 1=1");', check=False)
|
||||
self.assertNotEqual(failed.returncode, 0)
|
||||
|
||||
def test_worker_content_error_becomes_actionable_message(self):
|
||||
# Exercise PHP -> actual Python worker -> gateway -> PHP error mapping.
|
||||
out = self.root / 'mismatched-content'
|
||||
with Gateway(response={"title": "Gemeindefest 2099", "paragraphs": ["Der Aushang beschreibt ein Fest."]}):
|
||||
code = """try { contentpanel_render_news($argv[2], $argv[3], '2030-12-31-Test', 'template'); }
|
||||
catch (InvalidArgumentException $e) { echo $e->getMessage(); }"""
|
||||
result = self.php(code, str(self.pdf), str(out))
|
||||
self.assertIn('Die KI-Ausgabe enthält eine nicht belegte Zahl oder Kontaktangabe', result.stdout)
|
||||
self.assertNotIn('/opt/', result.stdout)
|
||||
self.assertIn('LLM-Ausgabe enthält eine nicht belegte Zahl oder Kontaktangabe', result.stderr)
|
||||
self.assertFalse(list(out.glob('*.png')))
|
||||
|
||||
def test_leitgedanken_parser_handles_hyphenation_and_duplicates(self):
|
||||
text = 'Sonntag 13.09.2030\n\nDie Ge-\nmeinde freut sich. Joh 3,16\n\nSonntag 13.09.2030\n\nDuplikat bleibt unberücksichtigt. Joh 3,16'
|
||||
result = self.php('echo json_encode(contentpanel_parse_leitgedanken($argv[2]));', text)
|
||||
self.assertEqual(json.loads(result.stdout), [{'bibeldatum': '2030-09-13', 'bibelwort': 'Joh 3,16', 'bibelvers': 'Die Gemeinde freut sich.'}])
|
||||
text = 'Sonntag 13./14.09.2030\nErster Vers. Joh 3,16\nWeiterer Vers. Ps 1,1\nSonntag 20.09.2030\n\nZweiter Vers. 1. Kor 13,1-3'
|
||||
result = self.php('echo json_encode(contentpanel_parse_leitgedanken($argv[2]));', text)
|
||||
self.assertEqual([item['bibelwort'] for item in json.loads(result.stdout)], ['Joh 3,16', '1. Kor 13,1-3'])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main(verbosity=2)
|
||||
@@ -0,0 +1,37 @@
|
||||
<?php
|
||||
// Only copied into a temporary test document root. Never used by the application.
|
||||
session_start();
|
||||
$groups = ['Allgemein' => !isset($_GET['deny'])];
|
||||
$user = ['mail' => 'member@example.test', 'first_name' => 'Test', 'last_name' => 'Person'];
|
||||
class TestStatement extends PDOStatement
|
||||
{
|
||||
public function __construct(private string $sql) {}
|
||||
public function execute(?array $params = null): bool
|
||||
{
|
||||
if (!str_starts_with($this->sql, 'SELECT')) {
|
||||
file_put_contents($_SERVER['DOCUMENT_ROOT'] . '/writes.jsonl', json_encode([$this->sql, $params]) . "\n", FILE_APPEND);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
public function fetchAll(int $mode = PDO::FETCH_DEFAULT, mixed ...$args): array
|
||||
{
|
||||
return str_contains($this->sql, 'FROM gottesdienst')
|
||||
? [['id' => 1, 'title' => 'Joh 3,16', 'subtitle' => '<script>unsafe()</script>', 'date' => '2030-12-31']]
|
||||
: [['id' => 2, 'title' => 'Gemeindefest', 'subtitle' => '', 'location' => 'Gemeindesaal',
|
||||
'time' => '11:30:00', 'date' => '2030-12-31', 'end_date' => null]];
|
||||
}
|
||||
}
|
||||
class TestPDO extends PDO
|
||||
{
|
||||
public function __construct() {}
|
||||
public function prepare(string $query, array $options = []): PDOStatement|false
|
||||
{
|
||||
return new TestStatement($query);
|
||||
}
|
||||
public function query(string $query, ?int $fetchMode = null, mixed ...$fetchModeArgs): PDOStatement|false
|
||||
{
|
||||
return new TestStatement($query);
|
||||
}
|
||||
public function exec(string $statement): int|false { return 0; }
|
||||
}
|
||||
$pdo = new TestPDO();
|
||||
Reference in New Issue
Block a user