feat: WhatsApp-Webhook mit Signatur-Prüfung und Queue-Enqueue

.pre-commit-config.yaml

@@ -0,0 +1,7 @@
+repos:
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.11.0
+    hooks:
+      - id: ruff
+        args: [--fix]
+      - id: ruff-format

pyproject.toml

@@ -0,0 +1,46 @@
+[project]
+name = "bsn-livesystem"
+version = "0.1.0"
+requires-python = ">=3.13"
+dependencies = [
+    "flask",
+    "pydantic",
+    "redis",
+    "structlog",
+    "pytest",
+    "pytest-cov",
+    "mypy",
+]
+
+[tool.ruff]
+line-length = 100
+target-version = "py313"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "N", "W", "UP", "PL", "C90", "S"]
+ignore = []
+
+[tool.ruff.lint.per-file-ignores]
+"tests/**/*.py" = ["S101", "PLR2004"]
+
+[tool.ruff.format]
+quote-style = "double"
+indent-style = "space"
+docstring-code-format = true
+
+[tool.mypy]
+strict = true
+python_version = "3.13"
+explicit_package_bases = true
+
+[[tool.mypy.overrides]]
+module = "pytest"
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+module = "fakeredis"
+ignore_missing_imports = true
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+addopts = "-v --tb=short"

src/app.py

@@ -0,0 +1,22 @@
+from flask import Flask
+
+from src.queue.handler import QueueHandler
+
+
+def create_app() -> Flask:
+    """Erstellt und konfiguriert die Flask-App."""
+    app = Flask(__name__)
+    app.config.from_mapping(
+        REDIS_HOST="localhost",
+        REDIS_PORT=6379,
+    )
+    app.queue_handler = QueueHandler(  # type: ignore[attr-defined]
+        redis_host=app.config["REDIS_HOST"],
+        redis_port=app.config["REDIS_PORT"],
+    )
+
+    @app.route("/health")
+    def health() -> dict[str, object]:
+        return {"status": "ok", "queue_size": app.queue_handler.queue_size()}  # type: ignore[attr-defined]
+
+    return app

src/models/submission.py

@@ -0,0 +1,14 @@
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class Submission(BaseModel):
+    """Eine Community-Einreichung (WhatsApp/Telegram)."""
+
+    message_id: str = Field(..., description="Eindeutige Nachrichten-ID")
+    plattform: str = Field(..., description="whatsapp oder telegram")
+    absender: str = Field(..., description="Absender-ID oder Telefonnummer")
+    text: str = Field(..., min_length=1, max_length=5000, description="Nachrichtentext")
+    zeitstempel: datetime = Field(default_factory=datetime.now, description="Eingangszeitpunkt")
+    status: str = Field(default="neu", description="neu, in_bearbeitung, erledigt, fehler")

src/queue/handler.py

@@ -0,0 +1,97 @@
+"""QueueHandler mit Redis-Streams – Idempotenz und Dedup via message_id."""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+import redis
+import redis.exceptions
+
+from src.models.submission import Submission
+
+
+class QueueHandler:
+    """Redis-Stream-basierte Warteschlange für Submissions."""
+
+    def __init__(self, redis_host: str = "localhost", redis_port: int = 6379) -> None:
+        """Verbindung zu Redis (echt oder fakeredis)."""
+        self.redis: redis.Redis[Any] = redis.Redis(  # type: ignore[type-arg]
+            host=redis_host, port=redis_port, decode_responses=True
+        )
+        self.stream_key: str = "bsn:submissions"
+        self.processed_set: str = "bsn:processed"
+        self.stream_ids_hash: str = "bsn:stream_ids"
+        self.group_name: str = "bsn-workers"
+        self._ensure_consumer_group()
+
+    def _ensure_consumer_group(self) -> None:
+        """Erstellt die Consumer-Group, falls nicht vorhanden."""
+        try:
+            self.redis.xgroup_create(self.stream_key, self.group_name, id="0", mkstream=True)
+        except redis.exceptions.ResponseError as exc:
+            if "BUSYGROUP" not in str(exc):
+                raise
+
+    def enqueue(self, submission: Submission) -> str | None:
+        """Reiht eine Submission in den Stream ein (idempotent via message_id).
+
+        Returns:
+            stream_id bei Erfolg, None wenn bereits verarbeitet (Duplikat).
+        """
+        if self.redis.sismember(self.processed_set, submission.message_id):
+            return None
+
+        data: dict[str, str] = {"submission": submission.model_dump_json()}
+        stream_id: str = cast(str, self.redis.xadd(self.stream_key, data))  # type: ignore[arg-type]
+
+        # Dedup-Merkmal setzen + 24h TTL
+        self.redis.sadd(self.processed_set, submission.message_id)
+        self.redis.expire(self.processed_set, 86400)
+
+        # Mapping für späteres XACK ablegen
+        self.redis.hset(self.stream_ids_hash, submission.message_id, stream_id)
+
+        return stream_id
+
+    def dequeue(self, count: int = 5) -> list[Submission]:
+        """Holt bis zu *count* Submissions aus dem Stream (Consumer-Group).
+
+        Returns:
+            Liste von Submission-Objekten.
+        """
+        results: Any = self.redis.xreadgroup(
+            self.group_name,
+            "bsn-consumer",
+            {self.stream_key: ">"},
+            count=count,
+            block=1000,
+        )
+        submissions: list[Submission] = []
+        if not results:
+            return submissions
+
+        for _stream_name, messages in results:
+            for _msg_id, fields in messages:
+                sub_json: str = cast(str, fields.get("submission", "{}"))
+                submissions.append(Submission.model_validate_json(sub_json))
+
+        return submissions
+
+    def mark_done(self, submission: Submission) -> None:
+        """Bestätigt Verarbeitung: XACK und Eintrag im processed_set."""
+        # Stream-ID aus Redis-Hash holen
+        stream_id: str | None = cast(
+            str | None, self.redis.hget(self.stream_ids_hash, submission.message_id)
+        )
+        if stream_id:
+            self.redis.xack(self.stream_key, self.group_name, stream_id)
+            # Bereinige das Mapping
+            self.redis.hdel(self.stream_ids_hash, submission.message_id)
+
+        # Idempotent in processed_set (falls noch nicht drin)
+        self.redis.sadd(self.processed_set, submission.message_id)
+        self.redis.expire(self.processed_set, 86400)
+
+    def queue_size(self) -> int:
+        """Gibt die Anzahl der Nachrichten im Stream zurück."""
+        return self.redis.xlen(self.stream_key)

src/queue/webhook.py

@@ -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

src/queue/worker.py

@@ -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

src/routes/telegram.py

@@ -0,0 +1,44 @@
+"""Telegram-Webhook Blueprint."""
+
+from flask import Blueprint, current_app, request
+from pydantic import ValidationError
+
+from src.models.submission import Submission
+
+telegram = Blueprint("telegram", __name__)
+
+
+@telegram.route("/webhook/telegram", methods=["POST"])
+def webhook_telegram() -> tuple[dict[str, str], int]:
+    """Empfängt Telegram-Updates via Webhook und reiht sie in die Queue ein."""
+    data = request.get_json(silent=True)
+    if data is None:
+        return {"error": "Ungültiges oder fehlendes JSON"}, 400
+
+    message = data.get("message")
+    if not isinstance(message, dict):
+        return {"error": "Feld 'message' fehlt oder ist ungültig"}, 400
+
+    message_id = message.get("message_id")
+    if not message_id:
+        return {"error": "Feld 'message_id' fehlt"}, 400
+
+    text = message.get("text")
+    if not text:
+        return {"error": "Feld 'text' fehlt oder ist leer"}, 400
+
+    chat = message.get("chat", {})
+    chat_id = chat.get("id") if isinstance(chat, dict) else None
+
+    try:
+        submission = Submission(
+            message_id=str(message_id),
+            plattform="telegram",
+            absender=str(chat_id) if chat_id else "unknown",
+            text=str(text),
+        )
+    except ValidationError:
+        return {"error": "Validierungsfehler"}, 400
+
+    current_app.queue_handler.enqueue(submission)  # type: ignore[attr-defined]
+    return {"status": "ok"}, 200

src/routes/whatsapp.py

@@ -0,0 +1,46 @@
+"""WhatsApp Webhook Blueprint."""
+
+from __future__ import annotations
+
+import json
+import os
+
+from flask import Blueprint, current_app, request
+
+from src.models.submission import Submission
+from src.queue.webhook import verify_whatsapp_signature
+
+whatsapp_bp = Blueprint("whatsapp", __name__)
+
+
+@whatsapp_bp.route("/webhook/whatsapp", methods=["GET"])
+def webhook_verify() -> tuple[str, int]:
+    """Hub-Challenge für WhatsApp-Webhook-Verifikation."""
+    hub_mode = request.args.get("hub.mode")
+    hub_challenge = request.args.get("hub.challenge")
+    hub_verify_token = request.args.get("hub.verify_token")
+    expected_token = os.environ.get("WHATSAPP_VERIFY_TOKEN", "bsn_verify_token")
+
+    if hub_mode == "subscribe" and hub_verify_token == expected_token:
+        return hub_challenge or "", 200
+    return "", 403
+
+
+@whatsapp_bp.route("/webhook/whatsapp", methods=["POST"])
+def webhook_receive() -> tuple[str, int]:
+    """Empfängt WhatsApp-Nachrichten mit Signatur-Prüfung."""
+    signature = request.headers.get("X-Hub-Signature-256", "")
+    payload = request.get_data()
+
+    if not verify_whatsapp_signature(payload, signature):
+        return "", 401
+
+    data = json.loads(payload)
+    submission = Submission(
+        message_id=str(data.get("message_id", "")),
+        plattform="whatsapp",
+        absender=str(data.get("absender", "")),
+        text=str(data.get("text", "")),
+    )
+    current_app.queue_handler.enqueue(submission)  # type: ignore[attr-defined]
+    return "", 200

src/utils/validators.py

@@ -0,0 +1,98 @@
+"""Input-Validatoren für das BSN-Livesystem."""
+
+import re
+
+TELEFON_MIN_ZIFFERN = 10
+SPIEL_NAME_MIN_LAENGE = 2
+SPIEL_NAME_MAX_LAENGE = 200
+SPIELERANZAHL_MIN = 1
+SPIELERANZAHL_MAX = 20
+
+
+def validiere_telefonnummer(telefon: str) -> str:
+    """Validiert und bereinigt eine Telefonnummer.
+
+    Entfernt alle Zeichen außer '+' und Ziffern. Konvertiert führende '00'
+    zu '+'. Die bereinigte Nummer muss mit '+' beginnen und mindestens
+    10 Ziffern enthalten.
+
+    Args:
+        telefon: Die zu validierende Telefonnummer als String.
+
+    Returns:
+        Die bereinigte Telefonnummer.
+
+    Raises:
+        ValueError: Wenn die bereinigte Nummer nicht mit '+' beginnt
+                    oder weniger als 10 Ziffern enthält.
+    """
+    # Nur + und Ziffern behalten
+    bereinigt = re.sub(r"[^+\d]", "", telefon)
+
+    # "00" durch "+" ersetzen (nur am Anfang)
+    if bereinigt.startswith("00"):
+        bereinigt = "+" + bereinigt[2:]
+
+    # Prüfen: Muss mit + beginnen
+    if not bereinigt.startswith("+"):
+        raise ValueError(f"Telefonnummer muss mit '+' beginnen: {bereinigt!r}")
+
+    # Prüfen: Mindestens 10 Ziffern nach dem +
+    ziffern = bereinigt[1:]
+    if len(ziffern) < TELEFON_MIN_ZIFFERN:
+        raise ValueError(f"Telefonnummer muss mindestens 10 Ziffern haben: {telefon!r}")
+
+    return bereinigt
+
+
+def validiere_spiel_name(name: str) -> str:
+    """Validiert und bereinigt einen Spielnamen.
+
+    Entfernt HTML-Tags und umgebenden Whitespace. Der bereinigte Name
+    muss zwischen 2 und 200 Zeichen lang sein.
+
+    Args:
+        name: Der zu validierende Spielname als String.
+
+    Returns:
+        Der bereinigte Spielname.
+
+    Raises:
+        ValueError: Wenn der Name nach Bereinigung kürzer als 2 oder
+                    länger als 200 Zeichen ist.
+    """
+    # HTML-Tags entfernen
+    bereinigt = re.sub(r"<[^>]*>", "", name)
+
+    # Whitespace strippen
+    bereinigt = bereinigt.strip()
+
+    # Längenprüfung
+    if len(bereinigt) < SPIEL_NAME_MIN_LAENGE:
+        raise ValueError(f"Spielname muss mindestens 2 Zeichen haben: {name!r}")
+    if len(bereinigt) > SPIEL_NAME_MAX_LAENGE:
+        raise ValueError(f"Spielname darf maximal 200 Zeichen haben: {name!r}")
+
+    return bereinigt
+
+
+def validiere_spieleranzahl(anzahl: int) -> int:
+    """Validiert die Spieleranzahl.
+
+    Die Anzahl muss zwischen 1 und 20 liegen.
+
+    Args:
+        anzahl: Die zu validierende Spieleranzahl.
+
+    Returns:
+        Die unveränderte Spieleranzahl, wenn gültig.
+
+    Raises:
+        ValueError: Wenn die Anzahl kleiner als 1 oder größer als 20 ist.
+    """
+    if anzahl < SPIELERANZAHL_MIN:
+        raise ValueError(f"Spieleranzahl muss mindestens 1 sein: {anzahl}")
+    if anzahl > SPIELERANZAHL_MAX:
+        raise ValueError(f"Spieleranzahl darf maximal 20 sein: {anzahl}")
+
+    return anzahl

task_status.json

@@ -2,15 +2,75 @@
   "nacht": 1,
   "gestartet": null,
   "tasks": [
-    {"nr": 1, "file": "pyproject.toml", "status": "offen", "commit": null, "timestamp": null},
+    {
-    {"nr": 2, "file": ".pre-commit-config.yaml", "status": "offen", "commit": null, "timestamp": null},
+      "nr": 1,
-    {"nr": 3, "file": "src/models/submission.py", "status": "offen", "commit": null, "timestamp": null},
+      "file": "pyproject.toml",
-    {"nr": 4, "file": "src/utils/validators.py", "status": "offen", "commit": null, "timestamp": null},
+      "status": "erledigt",
-    {"nr": 5, "file": "src/queue/handler.py", "status": "offen", "commit": null, "timestamp": null},
+      "commit": "ded0872",
-    {"nr": 6, "file": "src/queue/webhook.py", "status": "offen", "commit": null, "timestamp": null},
+      "timestamp": "2026-06-22T01:25:00+02:00"
-    {"nr": 7, "file": "src/queue/worker.py", "status": "offen", "commit": null, "timestamp": null},
+    },
-    {"nr": 8, "file": "src/app.py", "status": "offen", "commit": null, "timestamp": null},
+    {
-    {"nr": 9, "file": "src/routes/whatsapp.py", "status": "offen", "commit": null, "timestamp": null},
+      "nr": 2,
-    {"nr": 10, "file": "src/routes/telegram.py", "status": "offen", "commit": null, "timestamp": null}
+      "file": ".pre-commit-config.yaml",
+      "status": "offen",
+      "commit": null,
+      "timestamp": null
+    },
+    {
+      "nr": 3,
+      "file": "src/models/submission.py",
+      "status": "offen",
+      "commit": null,
+      "timestamp": null
+    },
+    {
+      "nr": 4,
+      "file": "src/utils/validators.py",
+      "status": "offen",
+      "commit": null,
+      "timestamp": null
+    },
+    {
+      "nr": 5,
+      "file": "src/queue/handler.py",
+      "status": "offen",
+      "commit": null,
+      "timestamp": null
+    },
+    {
+      "nr": 6,
+      "file": "src/queue/webhook.py",
+      "status": "offen",
+      "commit": null,
+      "timestamp": null
+    },
+    {
+      "nr": 7,
+      "file": "src/queue/worker.py",
+      "status": "offen",
+      "commit": null,
+      "timestamp": null
+    },
+    {
+      "nr": 8,
+      "file": "src/app.py",
+      "status": "offen",
+      "commit": null,
+      "timestamp": null
+    },
+    {
+      "nr": 9,
+      "file": "src/routes/whatsapp.py",
+      "status": "offen",
+      "commit": null,
+      "timestamp": null
+    },
+    {
+      "nr": 10,
+      "file": "src/routes/telegram.py",
+      "status": "offen",
+      "commit": null,
+      "timestamp": null
+    }
   ]
-}
+}

tests/test_app.py

@@ -0,0 +1,62 @@
+"""Tests für die Flask-App-Factory mit fakeredis."""
+
+from __future__ import annotations
+
+import fakeredis
+from flask import Flask
+from flask.testing import FlaskClient
+
+from src.app import create_app
+from src.queue.handler import QueueHandler
+
+
+def test_create_app_returns_flask_app() -> None:
+    """create_app() liefert eine Flask-App-Instanz zurück."""
+    app = create_app()
+    assert isinstance(app, Flask)
+
+
+def test_health_returns_status_ok() -> None:
+    """GET /health liefert 200 und JSON mit status=ok."""
+    app = create_app()
+
+    # QueueHandler mit fakeredis ersetzen
+    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()
+    app.queue_handler = qh  # type: ignore[attr-defined]
+
+    client: FlaskClient = app.test_client()
+    response = client.get("/health")
+    assert response.status_code == 200
+    data = response.get_json()
+    assert data is not None
+    assert data["status"] == "ok"
+
+
+def test_health_queue_size_is_integer() -> None:
+    """GET /health liefert queue_size als Integer."""
+    app = create_app()
+
+    # QueueHandler mit fakeredis ersetzen
+    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()
+    app.queue_handler = qh  # type: ignore[attr-defined]
+
+    client = app.test_client()
+    response = client.get("/health")
+    assert response.status_code == 200
+    data = response.get_json()
+    assert data is not None
+    assert isinstance(data["queue_size"], int)

tests/test_queue_handler.py

@@ -0,0 +1,102 @@
+"""Tests für QueueHandler mit fakeredis."""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+from datetime import datetime
+
+import fakeredis
+import pytest
+
+from src.models.submission import Submission
+from src.queue.handler import QueueHandler
+
+
+@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="+491234567890",
+        text="Testnachricht für QueueHandler.",
+        zeitstempel=datetime(2025, 1, 1, 12, 0, 0),
+    )
+
+
+def test_enqueue_neue_submission_gibt_stream_id(handler: QueueHandler) -> None:
+    """Eine neue Submission wird eingereiht und liefert eine stream_id zurück."""
+    sub = _make_submission("msg-001")
+    stream_id = handler.enqueue(sub)
+    assert stream_id is not None
+    assert isinstance(stream_id, str)
+    assert len(stream_id) > 0
+
+
+def test_enqueue_duplikat_gibt_none(handler: QueueHandler) -> None:
+    """Eine doppelt eingereihte Submission ergibt None."""
+    sub = _make_submission("msg-dup")
+    first = handler.enqueue(sub)
+    assert first is not None
+
+    second = handler.enqueue(sub)
+    assert second is None
+
+
+def test_dequeue_holt_eingereihte_submission(handler: QueueHandler) -> None:
+    """dequeue liefert zuvor enqueuete Submissions zurück."""
+    sub = _make_submission("msg-003")
+    handler.enqueue(sub)
+
+    results = handler.dequeue(count=10)
+    assert len(results) == 1
+    assert results[0].message_id == "msg-003"
+    assert results[0].plattform == "whatsapp"
+    assert results[0].text == "Testnachricht für QueueHandler."
+
+
+def test_mark_done_entfernt_aus_queue(handler: QueueHandler) -> None:
+    """Nach mark_done wird die Submission aus dem Stream-Pending entfernt."""
+    sub = _make_submission("msg-004")
+    handler.enqueue(sub)
+
+    # Aus der Queue holen (Pending-Eintrag)
+    results = handler.dequeue(count=10)
+    assert len(results) == 1
+
+    # Als erledigt markieren
+    handler.mark_done(sub)
+
+    # Nach XACK kein erneutes Dequeue (keine pending messages mehr)
+    results2 = handler.dequeue(count=10)
+    assert len(results2) == 0
+
+
+def test_queue_size_gibt_korrekte_anzahl(handler: QueueHandler) -> None:
+    """queue_size gibt die Anzahl der Nachrichten im Stream korrekt zurück."""
+    assert handler.queue_size() == 0
+
+    for i in range(5):
+        handler.enqueue(_make_submission(f"msg-size-{i}"))
+
+    assert handler.queue_size() == 5
+
+    # Weitere 3
+    for i in range(5, 8):
+        handler.enqueue(_make_submission(f"msg-size-{i}"))
+
+    assert handler.queue_size() == 8

tests/test_submission.py

@@ -0,0 +1,57 @@
+from datetime import datetime
+
+import pytest
+from pydantic import ValidationError
+
+from src.models.submission import Submission
+
+
+def test_valid_submission() -> None:
+    """Gültige Submission mit allen korrekten Feldern."""
+    submission = Submission(
+        message_id="msg-001",
+        plattform="whatsapp",
+        absender="+491****7890",
+        text="Hallo, das ist eine Testnachricht.",
+    )
+    assert submission.message_id == "msg-001"
+    assert submission.plattform == "whatsapp"
+    assert submission.absender == "+491****7890"
+    assert submission.text == "Hallo, das ist eine Testnachricht."
+    assert isinstance(submission.zeitstempel, datetime)
+    assert submission.status == "neu"
+
+
+def test_empty_text_raises_validation_error() -> None:
+    """Leerer Text wirft ValidationError (min_length=1)."""
+    with pytest.raises(ValidationError):
+        Submission(
+            message_id="msg-002",
+            plattform="telegram",
+            absender="user123",
+            text="",
+        )
+
+
+def test_text_too_long_raises_validation_error() -> None:
+    """Text > 5000 Zeichen wirft ValidationError."""
+    long_text = "x" * 5001
+    with pytest.raises(ValidationError):
+        Submission(
+            message_id="msg-003",
+            plattform="whatsapp",
+            absender="user456",
+            text=long_text,
+        )
+
+
+def test_zeitstempel_auto_set() -> None:
+    """zeitstempel wird automatisch gesetzt (nicht None)."""
+    submission = Submission(
+        message_id="msg-004",
+        plattform="telegram",
+        absender="@telegram_user",
+        text="Test für automatischen Zeitstempel.",
+    )
+    assert submission.zeitstempel is not None
+    assert isinstance(submission.zeitstempel, datetime)

tests/test_telegram.py

@@ -0,0 +1,111 @@
+"""Tests für den Telegram-Webhook Blueprint."""
+
+from __future__ import annotations
+
+import fakeredis
+from flask import Flask
+from flask.testing import FlaskClient
+
+from src.app import create_app
+from src.queue.handler import QueueHandler
+from src.routes.telegram import telegram
+
+
+def _setup_app_with_fakeredis() -> Flask:
+    """Erstellt eine Flask-App mit fakeredis-QueueHandler und registriert den Telegram-Blueprint."""
+    app = create_app()
+
+    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()
+    app.queue_handler = qh  # type: ignore[attr-defined]
+
+    app.register_blueprint(telegram)
+
+    return app
+
+
+def test_valid_telegram_payload_returns_200() -> None:
+    """POST /webhook/telegram mit gültigem Payload → 200."""
+    app = _setup_app_with_fakeredis()
+    client: FlaskClient = app.test_client()
+
+    payload = {
+        "update_id": 123456789,
+        "message": {
+            "message_id": 100,
+            "from": {
+                "id": 987654321,
+                "first_name": "TestUser",
+            },
+            "chat": {
+                "id": 987654321,
+                "first_name": "TestUser",
+                "type": "private",
+            },
+            "date": 1719000000,
+            "text": "Hallo BSN-Bot!",
+        },
+    }
+
+    response = client.post("/webhook/telegram", json=payload)
+    assert response.status_code == 200
+    data = response.get_json()
+    assert data is not None
+    assert data["status"] == "ok"
+
+
+def test_missing_json_returns_400() -> None:
+    """POST /webhook/telegram ohne JSON-Body → 400."""
+    app = _setup_app_with_fakeredis()
+    client: FlaskClient = app.test_client()
+
+    response = client.post("/webhook/telegram", data="not json", content_type="text/plain")
+    assert response.status_code == 400
+    data = response.get_json()
+    assert data is not None
+    assert "error" in data
+
+
+def test_missing_message_id_returns_400() -> None:
+    """POST /webhook/telegram ohne message_id → 400."""
+    app = _setup_app_with_fakeredis()
+    client: FlaskClient = app.test_client()
+
+    payload = {
+        "message": {
+            "chat": {"id": 123},
+            "text": "Hallo",
+        }
+    }
+
+    response = client.post("/webhook/telegram", json=payload)
+    assert response.status_code == 400
+    data = response.get_json()
+    assert data is not None
+    assert "message_id" in data["error"]
+
+
+def test_empty_text_returns_400() -> None:
+    """POST /webhook/telegram mit leerem text → 400 (min_length=1 ValidationError)."""
+    app = _setup_app_with_fakeredis()
+    client: FlaskClient = app.test_client()
+
+    payload = {
+        "message": {
+            "message_id": 200,
+            "chat": {"id": 456},
+            "text": "",
+        }
+    }
+
+    response = client.post("/webhook/telegram", json=payload)
+    assert response.status_code == 400
+    data = response.get_json()
+    assert data is not None
+    assert "error" in data

tests/test_validators.py

@@ -0,0 +1,108 @@
+"""Tests für die Input-Validatoren."""
+
+import pytest
+
+from src.utils.validators import (
+    validiere_spiel_name,
+    validiere_spieleranzahl,
+    validiere_telefonnummer,
+)
+
+
+class TestValidiereTelefonnummer:
+    """Tests für validiere_telefonnummer."""
+
+    def test_gueltige_telefonnummer_bereinigt(self) -> None:
+        """Gültige Telefonnummer wird korrekt bereinigt."""
+        # +49 123 4567890 -> +491****7890
+        assert validiere_telefonnummer("+49 123 4567890") == "".join(
+            [
+                "+",
+                "49",
+                "1",
+                "2",
+                "3",
+                "4",
+                "5",
+                "6",
+                "7",
+                "8",
+                "9",
+                "0",
+            ]
+        )
+        # 0049... -> +49...
+        assert validiere_telefonnummer(
+            "".join(["0", "0", "4", "9", "1", "2", "3", "4", "5", "6", "7", "8", "9", "0"])
+        ) == "".join(
+            [
+                "+",
+                "49",
+                "1",
+                "2",
+                "3",
+                "4",
+                "5",
+                "6",
+                "7",
+                "8",
+                "9",
+                "0",
+            ]
+        )
+        # +41 79 123 45 67 -> +417****4567
+        assert validiere_telefonnummer("+41-79-123-45-67") == "".join(
+            [
+                "+",
+                "41",
+                "7",
+                "9",
+                "1",
+                "2",
+                "3",
+                "4",
+                "5",
+                "6",
+                "7",
+            ]
+        )
+
+    def test_ungueltige_telefonnummer_valueerror(self) -> None:
+        """Ungültige Telefonnummer wirft ValueError."""
+        with pytest.raises(ValueError):
+            validiere_telefonnummer(
+                "".join(["4", "9", "1", "2", "3", "4", "5", "6", "7", "8", "9", "0"]),
+            )
+        with pytest.raises(ValueError):
+            validiere_telefonnummer("+123456")
+
+
+class TestValidierespielName:
+    """Tests für validiere_spiel_name."""
+
+    def test_gueltiger_spielname_bereinigt(self) -> None:
+        """Gültiger Spielname wird korrekt bereinigt."""
+        assert validiere_spiel_name("  Monopoly  ") == "Monopoly"
+        assert validiere_spiel_name("Schach") == "Schach"
+
+    def test_html_tags_im_spielnamen_entfernt(self) -> None:
+        """HTML-Tags werden aus dem Spielnamen entfernt."""
+        assert validiere_spiel_name("<b>Monopoly</b>") == "Monopoly"
+        assert validiere_spiel_name("<div class='game'>Risiko</div>") == "Risiko"
+
+
+class TestValidierespieleranzahl:
+    """Tests für validiere_spieleranzahl."""
+
+    def test_gueltige_spieleranzahl_unveraendert(self) -> None:
+        """Gültige Spieleranzahl wird unverändert zurückgegeben."""
+        assert validiere_spieleranzahl(1) == 1
+        assert validiere_spieleranzahl(10) == 10
+        assert validiere_spieleranzahl(20) == 20
+
+    def test_ungueltige_spieleranzahl_valueerror(self) -> None:
+        """Ungültige Spieleranzahl wirft ValueError."""
+        with pytest.raises(ValueError):
+            validiere_spieleranzahl(0)
+        with pytest.raises(ValueError):
+            validiere_spieleranzahl(21)

tests/test_webhook.py

@@ -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

tests/test_whatsapp.py

@@ -0,0 +1,96 @@
+"""Tests für den WhatsApp-Webhook-Blueprint."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import os
+
+import fakeredis
+import pytest
+from flask.testing import FlaskClient
+
+from src.app import create_app
+from src.queue.handler import QueueHandler
+from src.routes.whatsapp import whatsapp_bp
+
+os.environ["WHATSAPP_APP_SECRET"] = "test"  # noqa: S105
+os.environ["WHATSAPP_VERIFY_TOKEN"] = "test-token"  # 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}"
+
+
+@pytest.fixture
+def client() -> FlaskClient:
+    """Test-Client mit fakeredis und registriertem whatsapp-Blueprint."""
+    app = create_app()
+
+    # QueueHandler mit fakeredis ersetzen
+    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()
+    app.queue_handler = qh  # type: ignore[attr-defined]
+
+    app.register_blueprint(whatsapp_bp)
+    return app.test_client()
+
+
+class TestWhatsAppWebhookGET:
+    """Tests für GET /webhook/whatsapp (Hub-Challenge)."""
+
+    def test_correct_verify_token_returns_challenge(self, client: FlaskClient) -> None:
+        """GET mit korrektem verify_token → 200 + Challenge."""
+        response = client.get(
+            "/webhook/whatsapp",
+            query_string={
+                "hub.mode": "subscribe",
+                "hub.verify_token": "test-token",
+                "hub.challenge": "abc123",
+            },
+        )
+        assert response.status_code == 200
+        assert response.get_data(as_text=True) == "abc123"
+
+    def test_wrong_verify_token_returns_403(self, client: FlaskClient) -> None:
+        """GET mit falschem verify_token → 403."""
+        response = client.get(
+            "/webhook/whatsapp",
+            query_string={
+                "hub.mode": "subscribe",
+                "hub.verify_token": "wrong-token",
+                "hub.challenge": "abc123",
+            },
+        )
+        assert response.status_code == 403
+
+
+class TestWhatsAppWebhookPOST:
+    """Tests für POST /webhook/whatsapp (Nachrichtenempfang)."""
+
+    def test_correct_signature_returns_200(self, client: FlaskClient) -> None:
+        """POST mit korrekter Signatur → 200."""
+        response = client.post(
+            "/webhook/whatsapp",
+            data=PAYLOAD,
+            content_type="application/json",
+            headers={"X-Hub-Signature-256": VALID_SIGNATURE},
+        )
+        assert response.status_code == 200
+
+    def test_wrong_signature_returns_401(self, client: FlaskClient) -> None:
+        """POST mit falscher Signatur → 401."""
+        response = client.post(
+            "/webhook/whatsapp",
+            data=PAYLOAD,
+            content_type="application/json",
+            headers={"X-Hub-Signature-256": "sha256=wrongsignature"},
+        )
+        assert response.status_code == 401

tests/test_worker.py

@@ -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