224 lines
11 KiB
Python
224 lines
11 KiB
Python
"""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)
|