feat: Queue-Worker mit Batch-Verarbeitung und Fehlertoleranz

2026-06-22 06:34:40 +02:00
parent 3e360911f5
commit e382201878
4 changed files with 226 additions and 0 deletions
@@ -0,0 +1,49 @@
+import hashlib
+import hmac
+import json
+import os
+
+from src.models.submission import Submission
+from src.queue.handler import QueueHandler
+
+_MIN_TELEGRAM_TOKEN_LENGTH = 21
+
+
+def verify_whatsapp_signature(
+    payload: bytes, signature: str, app_secret: str | None = None
+) -> bool:
+    """Verifiziert die WhatsApp X-Hub-Signature-256."""
+    secret = app_secret or os.environ.get("WHATSAPP_APP_SECRET", "")
+    if not secret:
+        return False
+    expected = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
+    return hmac.compare_digest(f"sha256={expected}", signature)
+
+
+def verify_telegram_signature(token: str | None = None) -> bool:
+    """Verifiziert den Telegram Bot-Token."""
+    expected = token or os.environ.get("TELEGRAM_BOT_TOKEN", "")
+    if not expected:
+        return False
+    return len(expected) >= _MIN_TELEGRAM_TOKEN_LENGTH
+
+
+def validate_and_enqueue(payload: bytes, signature: str, plattform: str, qh: QueueHandler) -> bool:
+    """Validiert Signatur und legt in Queue ab."""
+    if plattform == "whatsapp":
+        if not verify_whatsapp_signature(payload, signature):
+            return False
+    elif plattform == "telegram":
+        if not verify_telegram_signature():
+            return False
+    else:
+        return False
+    # Submission bauen und enqueuen
+    data = json.loads(payload)
+    submission = Submission(
+        message_id=str(data.get("message_id", "")),
+        plattform=plattform,
+        absender=str(data.get("absender", "")),
+        text=str(data.get("text", "")),
+    )
+    return qh.enqueue(submission) is not None
@@ -0,0 +1,29 @@
+"""Queue-Worker: Pollt Submissions und verarbeitet sie in Batches."""
+
+from __future__ import annotations
+
+import time
+
+from src.models.submission import Submission
+from src.queue.handler import QueueHandler
+
+
+def verarbeite_submission(submission: Submission) -> bool:
+    """Verarbeitet eine einzelne Submission (Platzhalter für KI-Pipeline)."""
+    submission.status = "erledigt"
+    return True
+
+
+def worker_loop(qh: QueueHandler, poll_interval: int = 2, batch_size: int = 5) -> None:
+    """Endlos-Schleife: Pollt Queue, verarbeitet Batches."""
+    while True:
+        submissions = qh.dequeue(batch_size)
+        if not submissions:
+            time.sleep(poll_interval)
+            continue
+        for submission in submissions:
+            try:
+                if verarbeite_submission(submission):
+                    qh.mark_done(submission)
+            except Exception:  # noqa: S110
+                pass
@@ -0,0 +1,46 @@
+"""Tests für Webhook-Signatur-Prüfung."""
+
+import hashlib
+import hmac
+import os
+
+import pytest
+
+from src.queue.webhook import (
+    verify_telegram_signature,
+    verify_whatsapp_signature,
+)
+
+os.environ["WHATSAPP_APP_SECRET"] = "test"  # noqa: S105
+os.environ["TELEGRAM_BOT_TOKEN"] = "1234567890:ABCdefGHIjklMNOpqrsTUVwxyz"  # noqa: S105
+
+SECRET = "test"  # noqa: S105
+PAYLOAD = b'{"message_id": "123", "absender": "+491234", "text": "Hallo"}'
+EXPECTED_HMAC = hmac.new(SECRET.encode(), PAYLOAD, hashlib.sha256).hexdigest()
+VALID_SIGNATURE = f"sha256={EXPECTED_HMAC}"
+
+
+def test_correct_whatsapp_signature_returns_true() -> None:
+    """Korrekte WhatsApp-Signatur → True."""
+    assert verify_whatsapp_signature(PAYLOAD, VALID_SIGNATURE) is True
+
+
+def test_wrong_whatsapp_signature_returns_false() -> None:
+    """Falsche WhatsApp-Signatur → False."""
+    assert verify_whatsapp_signature(PAYLOAD, "sha256=wrongsignature") is False
+
+
+def test_missing_app_secret_returns_false(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Fehlender App-Secret → False (sicherer Fail)."""
+    monkeypatch.delenv("WHATSAPP_APP_SECRET", raising=False)
+    assert verify_whatsapp_signature(PAYLOAD, VALID_SIGNATURE) is False
+
+
+def test_correct_telegram_token_returns_true() -> None:
+    """Korrekter Telegram-Token → True."""
+    assert verify_telegram_signature() is True
+
+
+def test_invalid_telegram_token_too_short_returns_false() -> None:
+    """Ungültiger Telegram-Token (zu kurz) → False."""
+    assert verify_telegram_signature(token="short") is False  # noqa: S106
@@ -0,0 +1,102 @@
+"""Tests für Queue-Worker mit fakeredis."""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+from datetime import datetime
+from unittest.mock import patch
+
+import fakeredis
+import pytest
+
+from src.models.submission import Submission
+from src.queue.handler import QueueHandler
+from src.queue.worker import verarbeite_submission, worker_loop
+
+
+class BreakLoopError(Exception):
+    """Exception zum Durchbrechen der worker_loop-Endlosschleife in Tests."""
+
+
+@pytest.fixture
+def handler() -> Generator[QueueHandler]:
+    """QueueHandler mit fakeredis-Backend."""
+    fake_redis = fakeredis.FakeRedis(decode_responses=True)
+    qh = QueueHandler.__new__(QueueHandler)
+    qh.redis = fake_redis
+    qh.stream_key = "bsn:submissions"
+    qh.processed_set = "bsn:processed"
+    qh.stream_ids_hash = "bsn:stream_ids"
+    qh.group_name = "bsn-workers"
+    qh._ensure_consumer_group()
+    yield qh
+    fake_redis.flushall()
+
+
+def _make_submission(message_id: str = "msg-001") -> Submission:
+    """Erzeugt eine Test-Submission."""
+    return Submission(
+        message_id=message_id,
+        plattform="whatsapp",
+        absender="+491****7890",
+        text="Testnachricht für Queue-Worker.",
+        zeitstempel=datetime(2025, 1, 1, 12, 0, 0),
+    )
+
+
+def test_verarbeite_submission_setzt_status_auf_erledigt() -> None:
+    """verarbeite_submission setzt den Status auf 'erledigt'."""
+    sub = _make_submission("msg-001")
+    assert sub.status == "neu"
+
+    result = verarbeite_submission(sub)
+
+    assert result is True
+    assert sub.status == "erledigt"
+
+
+def test_worker_loop_verarbeitet_alle_submissions(handler: QueueHandler) -> None:
+    """worker_loop verarbeitet alle Submissions in der Queue."""
+    # Mehrere Submissions in die Queue einreihen
+    subs = [_make_submission(f"msg-w{i}") for i in range(3)]
+    for sub in subs:
+        handler.enqueue(sub)
+
+    # Verifiziere: Queue enthält 3 Einträge
+    assert handler.queue_size() == 3
+
+    # time.sleep so mocken, dass die Schleife nach einer Leer-Poll-Iteration abbricht
+    def _raise_on_sleep(seconds: int) -> None:  # noqa: ARG001
+        raise BreakLoopError
+
+    with patch("src.queue.worker.time.sleep", side_effect=_raise_on_sleep):
+        try:
+            worker_loop(handler, poll_interval=1, batch_size=5)
+        except BreakLoopError:
+            pass
+
+    # Nach Verarbeitung: dequeue liefert keine neuen Messages mehr
+    # (alle wurden per XACK bestätigt und sind nicht mehr "new")
+    remaining = handler.dequeue(count=10)
+    assert len(remaining) == 0, (
+        f"Erwartet 0 verbleibende Submissions, aber {len(remaining)} gefunden"
+    )
+
+
+def test_worker_loop_stoppt_bei_leerer_queue(handler: QueueHandler) -> None:
+    """worker_loop pollt leere Queue ohne Fehler (bricht nach erster sleep-Iteration ab)."""
+    call_count = 0
+
+    def _raise_after_first_sleep(seconds: int) -> None:  # noqa: ARG001
+        nonlocal call_count
+        call_count += 1
+        raise BreakLoopError
+
+    with patch("src.queue.worker.time.sleep", side_effect=_raise_after_first_sleep):
+        try:
+            worker_loop(handler, poll_interval=1, batch_size=5)
+        except BreakLoopError:
+            pass
+
+    # Sicherstellen, dass sleep genau einmal aufgerufen wurde (leere Queue → sleep)
+    assert call_count == 1