"""Tests für WhatsApp-Webhook-Blueprint.""" 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 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.""" response = self.client.get( "/webhook/whatsapp", query_string={ "hub.mode": "subscribe", "hub.verify_token": "bsn_verify_token", "hub.challenge": "12345", }, ) assert response.status_code == 200 assert response.data.decode() == "12345" def test_get_mit_falschem_verify_token(self) -> None: """GET mit falschem verify_token → 403.""" response = self.client.get( "/webhook/whatsapp", query_string={ "hub.mode": "subscribe", "hub.verify_token": "falscher_token", "hub.challenge": "12345", }, ) assert response.status_code == 403 def test_post_mit_korrekter_signatur(self) -> None: """POST mit korrekter Signatur → 200, Submission in Queue.""" wf_payload = _make_payload("491522123456", self._next_id(), "suche Catan") response = self._signed_post(wf_payload) 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 = {"object": "whatsapp", "entry": []} payload_bytes = json.dumps(payload).encode("utf-8") response = self.client.post( "/webhook/whatsapp", 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