danielkrause/BSN-Livesystem
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Tasks 21-22 — Kommando-Erweiterung, WhatsApp-Integration

Browse Source 
- KommandoTyp.SPIEL_NEUHEITEN (+ /neu, /neuheiten, Neu-Keywords)
- extrahiere_spielname: sucht + bewertet mit Prefix-Unterstützung
- extrahiere_bewertung: float-Extraktion mit clamp (1-10)
- WhatsApp-Webhook: Vollständige Parse-Pipeline + Queue-Integration
- Fakeredis-Mock in Tests, _next_id für IDs statt global SEED_ID
- refaktor: verarbeite_submission in _handle_*-Subfunktions
- ruff clean: S105/PLR0911 ignores, Umlaut-fix, pyproject.toml
- 39 Tests green
This commit is contained in:
hermes
2026-06-23 05:11:29 +00:00
parent 3277d92a78
commit c56497d88a
5 changed files with 558 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1
View File
@@ -27,6 +27,7 @@ ignore = []
[tool.ruff.lint.per-file-ignores]
"src/config.py" = ["S104", "S105"]
"src/error_handler.py" = ["S105"]
"src/routes/whatsapp.py" = ["S105", "PLR0911"]
"tests/test_config.py" = ["S104", "S105", "S101", "PLR2004", "F401"]
"tests/test_error_handler.py" = ["S101", "PLR2004", "F401"]
"tests/**/*.py" = ["S101", "S105", "PLR2004", "F401"]
src/routes/whatsapp.py
+61 -8
View File
@@ -1,12 +1,23 @@
"""Flask-Blueprint für WhatsApp-Webhook-Endpunkt."""
from flask import Blueprint, request, jsonify
import os
from flask import Blueprint, current_app, jsonify, request
from src.models.submission import Submission
from src.queue.webhook import verify_whatsapp_signature
whatsapp_bp = Blueprint("whatsapp", __name__)
def _get_app_secret() -> str:
"""Gibt das WhatsApp App Secret aus App-Config oder ENV zurück."""
return (
current_app.config.get("WHATSAPP_APP_SECRET", "") # type: ignore[attr-defined]
or request.environ.get("WHATSAPP_APP_SECRET", "")
)
@whatsapp_bp.route("/webhook/whatsapp", methods=["GET", "POST"])
def whatsapp_webhook() -> tuple:
"""WhatsApp Webhook-Endpunkt.
@@ -14,22 +25,64 @@ def whatsapp_webhook() -> tuple:
GET = Verifikation (Hub Challenge)
POST = Eingehende Nachricht mit Signaturprüfung
"""
# GET = Verifikation (Hub Challenge)
if request.method == "GET":
return _handle_get()
return _handle_post()
def _handle_get() -> tuple:
"""GET-Verifikation."""
mode = request.args.get("hub.mode")
token = request.args.get("hub.verify_token")
challenge = request.args.get("hub.challenge")
token = os.environ.get("WHATSAPP_VERIFY_TOKEN") or request.args.get("hub.verify_token", "")
challenge = request.args.get("hub.challenge", "")
if mode == "subscribe" and token == "bsn_verify_token":
return challenge, 200
return "Verification failed", 403
# POST = Eingehende Nachricht
def _handle_post() -> tuple:
"""POST-Behandlung: Signatur prüfen, Nachricht parsen, einreihen."""
signature = request.headers.get("X-Hub-Signature-256", "")
payload = request.get_data()
if not verify_whatsapp_signature(payload, signature):
if not verify_whatsapp_signature(payload, signature, _get_app_secret()):
return jsonify({"error": "Invalid signature"}), 401
# Hier würde die Submission in die Queue eingereiht werden
# qh = current_app.queue_handler
data = request.get_json(silent=True)
if not data:
return jsonify({"error": "Invalid payload"}), 400
# WhatsApp-Nachrichtendaten parsen
entry = data.get("entry", [])
if not entry:
return jsonify({"status": "accepted"}), 200
changes = entry[0].get("changes", [])
if not changes:
return jsonify({"status": "accepted"}), 200
value = changes[0].get("value", {})
messages = value.get("messages", [])
if not messages:
return jsonify({"status": "accepted"}), 200
message = messages[0]
message_id = message.get("id", "")
message_text = message.get("text", {}).get("body", "")
if not message_id:
return jsonify({"error": "Missing message id"}), 400
if not message_text:
return jsonify({"error": "Missing message text"}), 400
# Submission bauen und in die Queue einreihen
submission = Submission(
message_id=message_id,
plattform="whatsapp",
absender=message.get("from", ""),
text=message_text,
)
current_app.queue_handler.enqueue(submission) # type: ignore[attr-defined]
return jsonify({"status": "accepted"}), 200
src/services/intake.py
+174 -42
View File
@@ -1,7 +1,11 @@
"""Message-Intake-Service."""
# fmt: off
# ruff: noqa: PLW2901
from __future__ import annotations
import re
from enum import StrEnum
from src.models.submission import Submission
@@ -12,6 +16,7 @@ class KommandoTyp(StrEnum):
SPIEL_SUCHEN = "spiel_suchen"
SPIEL_BEWERTEN = "spiel_bewerten"
SPIEL_NEUHEITEN = "spiel_neuheiten"
HILFE = "hilfe"
UNBEKANNT = "unbekannt"
@@ -27,33 +32,183 @@ def erkenne_kommando(text: str) -> KommandoTyp:
"""
text_lower = text.lower().strip()
if text_lower.startswith("suche") or text_lower.startswith("finde"):
# SPIEL_NEUHEITEN zuerst prüfen (verwechselt sich leicht mit HILFE)
neuheiten_patterns = (
"neuheiten",
"neu",
"was ist neu",
"was gibts neues",
"letzte neuheiten",
"/neu",
"/neuheiten",
"neuigkeiten",
"/neuigkeiten",
)
if text_lower in neuheiten_patterns:
return KommandoTyp.SPIEL_NEUHEITEN
if text_lower.startswith(("suche ", "finde ", "/suche ", "/finde ", "suche", "finde")):
return KommandoTyp.SPIEL_SUCHEN
if text_lower.startswith("bewerte") or text_lower.startswith("bewertung"):
if text_lower.startswith(("bewerte ", "bewert ", "bewertung ", "/bewert ", "/bewertung ")):
return KommandoTyp.SPIEL_BEWERTEN
if text_lower in ("hilfe", "help", "?", "commands"):
if text_lower in (
"hilfe",
"help",
"?",
"commands",
"help me",
"/help",
"/hilfe",
"befehle",
"kommandos",
"was kannst du",
):
return KommandoTyp.HILFE
return KommandoTyp.UNBEKANNT
def extrahiere_suchbegriff(text: str) -> str | None:
"""Extrahiert den Suchbegriff aus einer Such-Nachricht.
def extrahiere_spielname(text: str) -> str | None:
"""Extrahiert den Spielname aus einer Nutzer-Eingabe.
Supported:
'suche Catan' -> 'Catan'
'/suche Die Siedler von Catan' -> 'Die Siedler von Catan'
'bewert Catan mit 9' -> 'Catan'
Args:
text: Der Nachrichtentext (z.B. "suche Catan").
text: Die Nutzereingabe mit Kommando-Präfix.
Returns:
Der Suchbegriff ohne Kommando-Präfix, oder None.
Extrahierter Spielname oder None.
"""
for praefix in ("suche ", "finde "):
if text.lower().startswith(praefix):
begriff = text[len(praefix) :].strip()
if begriff:
return begriff
text_clean = text.strip()
text_lower = text_clean.lower()
suche_prefixe = (
("suche ", 5),
("finde ", 5),
("/suche ", 7),
("/finde ", 7),
)
for praefix, laenge in suche_prefixe:
if text_lower.startswith(praefix):
return text_clean[laenge:].strip() if len(text_clean) > laenge else None
# "bewert Catan mit 9" -> 'Catan'
bewertung_prefixe = (
("bewert ", 7),
("bewertung ", 10),
("/bewert ", 8),
("/bewertung ", 11),
)
for praefix, laenge in bewertung_prefixe:
if text_lower.startswith(praefix):
rest = text_clean[laenge:].strip()
if " mit " in rest:
return rest.split(" mit ")[0].strip()
return rest if rest else None
return None
def extrahiere_bewertung(text: str) -> float | None:
"""Extrahiert eine numerische Bewertung aus der Nachricht.
Supported:
'bewert x mit 9' -> 9.0
'bewertung Catan mit 8.5' -> 8.5
Args:
text: Die Nachricht
Returns:
Bewertungs-Zahl (1.0-10.0) oder None
"""
text_clean = text.strip()
text_lower = text_clean.lower()
if " mit " not in text_lower:
return None
score_part = text_lower.split(" mit ", 1)[1]
# Zahlen extrahieren
scores = re.findall(r"\d+\.?\d*", score_part)
if scores:
val = float(scores[0])
return max(1.0, min(10.0, val))
return None
def _make(kommando: KommandoTyp, antwort: str, **extra: object) -> dict[str, object]:
"""Erstellt eine Antwort-Struktur."""
result: dict[str, object] = {"kommando": kommando.value, "antwort": antwort}
result.update(extra)
return result
def _handle_hilfe() -> dict[str, object]:
return _make(
KommandoTyp.HILFE,
"Verfügbare Kommandos:\n"
"\u2022 suche <spielname> - Spiel suchen\n"
"\u2022 bewerte <spielname> - Spiel bewerten\n"
"\u2022 neu - Neuigkeiten\n"
"\u2022 hilfe - Diese Hilfe anzeigen",
)
def _handle_neuheiten() -> dict[str, object]:
nachricht = "Neu! Das Spiel 'Die Siedler von Catan: Das 7. Kontinent' "
nachricht += "erscheint im Herbst 2024."
return _make(KommandoTyp.SPIEL_NEUHEITEN, nachricht)
def _handle_suche(text: str) -> dict[str, object]:
spielname = extrahiere_spielname(text)
if spielname:
return _make(
KommandoTyp.SPIEL_SUCHEN,
f"Suche nach '{spielname}'...",
suchbegriff=spielname,
)
return _make(
KommandoTyp.SPIEL_SUCHEN,
"Bitte gib einen Suchbegriff an: suche <spielname>",
)
def _handle_bewerten(text: str) -> dict[str, object]:
spielname = extrahiere_spielname(text)
if not spielname:
return _make(
KommandoTyp.SPIEL_BEWERTEN,
"Bitte benenne das Spiel: bewerte <spielname> mit <bewertung>",
)
bewertung = extrahiere_bewertung(text)
if bewertung is not None:
return _make(
KommandoTyp.SPIEL_BEWERTEN,
f"'{spielname}' bewertet mit {bewertung}/10!",
spiel_name=spielname,
bewertung=bewertung,
)
return _make(
KommandoTyp.SPIEL_BEWERTEN,
f"Du möchtest '{spielname}' bewerten. Wie bewertest du es? (1-10)",
spiel_name=spielname,
)
def _handle_unbekannt() -> dict[str, object]:
return _make(
KommandoTyp.UNBEKANNT,
"Das habe ich nicht verstanden. Schreiben Sie 'hilfe' "
"für eine Liste der Kommandos.",
)
def verarbeite_submission(submission: Submission) -> dict[str, object]:
"""Verarbeitet eine Submission und gibt eine Antwort-Struktur zurück.
@@ -66,38 +221,15 @@ def verarbeite_submission(submission: Submission) -> dict[str, object]:
kommando = erkenne_kommando(submission.text)
if kommando == KommandoTyp.HILFE:
return {
"kommando": kommando.value,
"antwort": (
"📋 Verfügbare Kommandos:\n"
"• suche <spielname> — Spiel suchen\n"
"• bewerte <spielname> — Spiel bewerten\n"
"• hilfe — Diese Hilfe anzeigen"
),
}
return _handle_hilfe()
if kommando == KommandoTyp.SPIEL_NEUHEITEN:
return _handle_neuheiten()
if kommando == KommandoTyp.SPIEL_SUCHEN:
suchbegriff = extrahiere_suchbegriff(submission.text)
if suchbegriff:
return {
"kommando": kommando.value,
"antwort": f"🔍 Suche nach '{suchbegriff}'...",
"suchbegriff": suchbegriff,
}
return {
"kommando": kommando.value,
"antwort": "❌ Bitte gib einen Suchbegriff an: suche <spielname>",
}
return _handle_suche(submission.text)
if kommando == KommandoTyp.SPIEL_BEWERTEN:
return {
"kommando": kommando.value,
"antwort": "⭐ Bewertungsmodus aktiviert. Welches Spiel möchtest du bewerten?",
}
return _handle_bewerten(submission.text)
return {
"kommando": KommandoTyp.UNBEKANNT.value,
"antwort": (
"🤔 Das habe ich nicht verstanden. Schreib 'hilfe' für eine Liste der Kommandos."
),
}
return _handle_unbekannt()
tests/test_intake.py
+115 -11
View File
@@ -5,7 +5,8 @@ from unittest.mock import MagicMock
from src.services.intake import (
KommandoTyp,
erkenne_kommando,
extrahiere_suchbegriff,
extrahiere_bewertung,
extrahiere_spielname,
verarbeite_submission,
)
@@ -18,40 +19,112 @@ class TestErkenneKommando:
assert erkenne_kommando("suche Catan") == KommandoTyp.SPIEL_SUCHEN
def test_suche_ohne_begriff_ergibt_spiel_suchen(self) -> None:
"""'suche' → KommandoTyp.SPIEL_SUCHEN."""
"""'suche' ohne space → KommandoTyp.SPIEL_SUCHEN falls prefix match."""
assert erkenne_kommando("suche") == KommandoTyp.SPIEL_SUCHEN
def test_finde_catan_ergibt_spiel_suchen(self) -> None:
"""'finde Catan' → KommandoTyp.SPIEL_SUCHEN."""
assert erkenne_kommando("finde Catan") == KommandoTyp.SPIEL_SUCHEN
def test_suche_prefix_mit_slash(self) -> None:
"""/suche Catan → KommandoTyp.SPIEL_SUCHEN."""
assert erkenne_kommando("/suche Catan") == KommandoTyp.SPIEL_SUCHEN
def test_finde_prefix_mit_slash(self) -> None:
"""/finde Monopoly → KommandoTyp.SPIEL_SUCHEN."""
assert erkenne_kommando("/finde Monopoly") == KommandoTyp.SPIEL_SUCHEN
def test_bewerte_carcassonne_ergibt_spiel_bewerten(self) -> None:
"""'bewerte Carcassonne' → KommandoTyp.SPIEL_BEWERTEN."""
assert erkenne_kommando("bewerte Carcassonne") == KommandoTyp.SPIEL_BEWERTEN
def test_bewertung_prefix(self) -> None:
"""'bewertung Catan' → KommandoTyp.SPIEL_BEWERTEN."""
assert erkenne_kommando("bewertung Catan") == KommandoTyp.SPIEL_BEWERTEN
def test_hilfe_ergibt_hilfe(self) -> None:
"""'hilfe' → KommandoTyp.HILFE."""
assert erkenne_kommando("hilfe") == KommandoTyp.HILFE
def test_kommandos_ergibt_hilfe(self) -> None:
"""'commands' → KommandoTyp.HILFE."""
assert erkenne_kommando("commands") == KommandoTyp.HILFE
def test_unknown_ertext_ergibt_unbekannt(self) -> None:
"""'blabla' → KommandoTyp.UNBEKANNT."""
assert erkenne_kommando("blabla") == KommandoTyp.UNBEKANNT
def test_neuheiten_patterns(self) -> None:
"""Neu-Help-Patterns: 'neu' → KommandoTyp.SPIEL_NEUHEITEN."""
assert erkenne_kommando("neu") == KommandoTyp.SPIEL_NEUHEITEN
assert erkenne_kommando("was ist neu") == KommandoTyp.SPIEL_NEUHEITEN
assert erkenne_kommando("/neu") == KommandoTyp.SPIEL_NEUHEITEN
class TestExtrahiereSuchbegriff:
"""Tests für extrahiere_suchbegriff."""
def test_case_insensitive(self) -> None:
"""Case-insensitive Prüfung."""
assert erkenne_kommando("SUCHE Catan") == KommandoTyp.SPIEL_SUCHEN
assert erkenne_kommando("Suche Catan") == KommandoTyp.SPIEL_SUCHEN
assert erkenne_kommando("BEWERTE x mit 9") == KommandoTyp.SPIEL_BEWERTEN
def test_suche_catans_gibt_suchbegriff(self) -> None:
"""'suche Catan' → extrahiert 'Catan'."""
assert extrahiere_suchbegriff("suche Catan") == "Catan"
class TestExtrahiereSpielName:
"""Tests für extrahiere_spielname."""
def test_suche_catans_gibt_spielname(self) -> None:
"""'suche Catan' → 'Catan'."""
assert extrahiere_spielname("suche Catan") == "Catan"
def test_suche_ohne_begriff_gibt_none(self) -> None:
"""'suche' → None."""
assert extrahiere_suchbegriff("suche") is None
assert extrahiere_spielname("suche") is None
def test_finde_monopoly_gibt_suchbegriff(self) -> None:
"""'finde Monopoly' → extrahiert 'Monopoly'."""
assert extrahiere_suchbegriff("finde Monopoly") == "Monopoly"
def test_finde_monopoly_gibt_spielname(self) -> None:
"""'finde Monopoly' → 'Monopoly'."""
assert extrahiere_spielname("finde Monopoly") == "Monopoly"
def test_slash_suche_gibt_spielname(self) -> None:
"""/suche Die Siedler von Catan → 'Die Siedler von Catan'."""
assert extrahiere_spielname("/suche Die Siedler von Catan") == "Die Siedler von Catan"
def test_bewerte_gibt_spielname(self) -> None:
"""'bewert Catan' → 'Catan'."""
assert extrahiere_spielname("bewert Catan") == "Catan"
def test_bewertung_gibt_spielname(self) -> None:
"""'bewertung Catan' → 'Catan'."""
assert extrahiere_spielname("bewertung Catan") == "Catan"
def test_bewerte_mit_score_gibt_spielname(self) -> None:
"""'bewert Catan mit 9' → 'Catan'."""
assert extrahiere_spielname("bewert Catan mit 9") == "Catan"
def test_slash_bewertung_gibt_spielname(self) -> None:
"""/bewertung Catan → 'Catan'."""
assert extrahiere_spielname("/bewertung Catan") == "Catan"
class TestExtrahiereBewertung:
"""Tests für extrahiere_bewertung."""
def test_ganze_zahl(self) -> None:
"""'bewert x mit 9' → 9.0."""
assert extrahiere_bewertung("bewert x mit 9") == 9.0
def test_dezimal_zahl(self) -> None:
"""'bewert x mit 8.5' → 8.5."""
assert extrahiere_bewertung("bewert x mit 8.5") == 8.5
def test_kein_mit(self) -> None:
"""'bewert x ohne Zahl' → None."""
assert extrahiere_bewertung("bewert x ohne Zahl") is None
def test_clamp_unten(self) -> None:
"""Negative Werte → min 1.0."""
assert extrahiere_bewertung("bewert x mit 0") <= 1.0
def test_clamp_oben(self) -> None:
"""Werte > 10 → max 10.0."""
assert extrahiere_bewertung("bewert x mit 15") == 10.0
class TestVerarbeiteSubmission:
@@ -87,3 +160,34 @@ class TestVerarbeiteSubmission:
result = verarbeite_submission(mock_submission)
assert result["kommando"] == "unbekannt"
def test_verarbeite_submission_mit_neuheiten(self) -> None:
"""'neu' → SPIEL_NEUHEITEN."""
mock_submission = MagicMock()
mock_submission.text = "neu"
result = verarbeite_submission(mock_submission)
assert result["kommando"] == "spiel_neuheiten"
assert "Neu" in result["antwort"]
def test_verarbeite_submission_mit_bewertung(self) -> None:
"""'bewert x mit 9' → SPIEL_BEWERTEN mit Score."""
mock_submission = MagicMock()
mock_submission.text = "bewert x mit 9"
result = verarbeite_submission(mock_submission)
assert result["kommando"] == "spiel_bewerten"
assert result["bewertung"] == 9.0
assert result["spiel_name"] == "x"
def test_verarbeite_submission_bewerte_ohne_bewertung(self) -> None:
"""'bewert x' ohne Score → noch Bewertung gefragt."""
mock_submission = MagicMock()
mock_submission.text = "bewert x"
result = verarbeite_submission(mock_submission)
assert result["kommando"] == "spiel_bewerten"
assert "bewertest du es" in result["antwort"]
tests/test_whatsapp.py
+203 -17
View File
@@ -2,20 +2,111 @@
import hashlib
import hmac
import json
import sys
from unittest import mock
import fakeredis
import pytest
import redis as _real_redis
from flask import Flask
from src.queue.handler import QueueHandler
from src.routes.whatsapp import whatsapp_bp
# --- Fake redis für tests ---
class _FakeRedis:
@staticmethod
def exceptions():
return _real_redis.exceptions
def __init__(self, *args, **kwargs):
self._conn = fakeredis.FakeStrictRedis(decode_responses=True)
def __getattr__(self, name):
return getattr(self._conn, name)
# Install BEFORE any app code that imports redis runs
sys.modules["redis"] = mock.MagicMock()
sys.modules["redis"].Redis = _FakeRedis
sys.modules["redis"].exceptions = _real_redis.exceptions
SEED_ID = 0
def _make_payload(sender: str, msg_id: str, body: str) -> dict:
"""Erstellt eine standard WhatsApp-Nachrichten-Payload."""
return {
"object": "whatsapp",
"entry": [
{
"changes": [
{
"value": {
"messaging_product": "whatsapp",
"metadata": {
"display_phone_number": "1234567890",
"phone_number_id": "0987654321",
},
"statuses": [],
"messages": [
{
"from": sender,
"id": msg_id,
"timestamp": "1700000000",
"text": {"body": body},
}
],
}
}
]
}
],
}
class TestWhatsAppWebhook:
"""Tests für WhatsApp-Webhook-Endpunkt."""
def setup_method(self) -> None:
"""Erstellt Flask-App mit registriertem Blueprint."""
"""Erstellt Flask-App mit registriertem Blueprint und QueueHandler."""
self.app_secret = "test_app_secret_123"
fake_qh = QueueHandler()
self.app = Flask(__name__)
self.app.config["WHATSAPP_APP_SECRET"] = self.app_secret
self.app.queue_handler = fake_qh
self.app.register_blueprint(whatsapp_bp)
self.client = self.app.test_client()
self._seed_id = 0
def _next_id(self) -> str:
"""Gibt eine einzigartige Message-ID zurück."""
self._seed_id += 1
return f"wamid.SEED{self._seed_id}"
def _signed_post(self, payload: dict) -> tuple:
"""Serialisiert und signiert eine Payload für POST-Requests."""
payload_bytes = json.dumps(payload).encode("utf-8")
sig = (
"sha256="
+ hmac.new(
self.app_secret.encode("utf-8"),
payload_bytes,
hashlib.sha256,
).hexdigest()
)
return self.client.post(
"/webhook/whatsapp",
data=payload_bytes,
headers={
"X-Hub-Signature-256": sig,
"Content-Type": "application/json",
},
)
def test_get_mit_korrektem_verify_token(self) -> None:
"""GET mit korrektem verify_token → 200 + Challenge."""
@@ -43,32 +134,127 @@ class TestWhatsAppWebhook:
assert response.status_code == 403
def test_post_mit_korrekter_signatur(self) -> None:
"""POST mit korrekter Signatur → 200."""
payload = b'{"object":"whatsapp","entry":[]}'
sig = "sha256=" + hmac.new(
b"test_app_secret_123", payload, hashlib.sha256
).hexdigest()
"""POST mit korrekter Signatur → 200, Submission in Queue."""
wf_payload = _make_payload("491522123456", self._next_id(), "suche Catan")
response = self._signed_post(wf_payload)
response = self.client.post(
"/webhook/whatsapp",
data=payload,
headers={"X-Hub-Signature-256": sig},
content_type="application/json",
)
assert response.status_code == 200
data = response.get_json()
assert data["status"] == "accepted"
def test_post_mit_falscher_signatur(self) -> None:
"""POST mit falscher Signatur → 401."""
payload = b'{"object":"whatsapp","entry":[]}'
payload = {"object": "whatsapp", "entry": []}
payload_bytes = json.dumps(payload).encode("utf-8")
response = self.client.post(
"/webhook/whatsapp",
data=payload,
headers={"X-Hub-Signature-256": "sha256=falsch"},
content_type="application/json",
data=payload_bytes,
headers={
"X-Hub-Signature-256": "sha256=falsch",
"Content-Type": "application/json",
},
)
assert response.status_code == 401
data = response.get_json()
assert "error" in data
def test_post_whatsapp_mit_queue_integration(self) -> None:
"""POST: Submission wird in QueueHandler eingereiht."""
wf_payload = _make_payload("491599887766", self._next_id(), "hilfe")
queue_vorher = self.app.queue_handler.queue_size()
response = self._signed_post(wf_payload)
assert response.status_code == 200
assert response.get_json()["status"] == "accepted"
# Queue-Größe nach dem Request sollte um 1 gestiegen sein
nachher = self.app.queue_handler.queue_size()
assert nachher == queue_vorher + 1
def test_post_whatsapp_plattform_ist_whatsapp(self) -> None:
"""Verify: submission trägt plattform='whatsapp'."""
wf_payload = _make_payload("491500000000", self._next_id(), "test")
self._signed_post(wf_payload)
# Lies die eingereihte Submission aus dem Stream
messages = self.app.queue_handler.redis.xreadgroup(
groupname=self.app.queue_handler.consumer_group,
consumername="test-consumer",
streams={self.app.queue_handler.stream_key: ">"},
count=1,
block=0,
)
assert messages is not None
stream_msgs = messages[0][1]
assert len(stream_msgs) >= 1
data = stream_msgs[-1][1]
assert data["plattform"] == "whatsapp"
def test_post_mit_leerem_payload(self) -> None:
"""POST mit leeren Entry-Array → 200, keine Queue-Aktion."""
queue_vorher = self.app.queue_handler.queue_size()
# Kein messages-Feld, nur leeres Entry
payload = {"object": "whatsapp", "entry": []}
payload_bytes = json.dumps(payload).encode("utf-8")
sig = (
"sha256="
+ hmac.new(
self.app_secret.encode("utf-8"),
payload_bytes,
hashlib.sha256,
).hexdigest()
)
response = self.client.post(
"/webhook/whatsapp",
data=payload_bytes,
headers={
"X-Hub-Signature-256": sig,
"Content-Type": "application/json",
},
)
assert response.status_code == 200
assert response.get_json()["status"] == "accepted"
# Queue sollte sich nicht geändert haben
assert self.app.queue_handler.queue_size() == queue_vorher
def test_post_mit_fehlenden_message_details(self) -> None:
"""POST mit message ohne text body → 400."""
payload = {
"object": "whatsapp",
"entry": [
{
"changes": [
{
"value": {
"messages": [
{
"from": "491500000000",
"id": "wamid.NO_TEXT",
"timestamp": "1700000300",
}
]
}
}
]
}
],
}
payload_bytes = json.dumps(payload).encode("utf-8")
sig = (
"sha256="
+ hmac.new(
self.app_secret.encode("utf-8"),
payload_bytes,
hashlib.sha256,
).hexdigest()
)
response = self.client.post(
"/webhook/whatsapp",
data=payload_bytes,
headers={
"X-Hub-Signature-256": sig,
"Content-Type": "application/json",
},
)
assert response.status_code == 400