feat: Webhooks (Aufgaben 6-10) — webhook.py, worker.py, Flask-App, WhatsApp + Telegram Routes + Tests

2026-06-22 07:58:25 +00:00
parent 6f7c5eada2
commit 56b02c1c2d
11 changed files with 629 additions and 0 deletions
@@ -0,0 +1,79 @@
+"""Tests für WhatsApp-Webhook-Blueprint."""
+
+import hashlib
+import hmac
+from unittest.mock import patch
+
+from flask import Flask
+
+from src.routes.whatsapp import whatsapp_bp
+from src.queue.webhook import verify_whatsapp_signature
+
+
+class TestWhatsAppWebhook:
+    """Tests für WhatsApp-Webhook-Endpunkt."""
+
+    def setup_method(self) -> None:
+        """Erstellt Flask-App mit registriertem Blueprint."""
+        self.app = Flask(__name__)
+        self.app.register_blueprint(whatsapp_bp)
+        self.client = self.app.test_client()
+
+    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."""
+        payload = b'{"object":"whatsapp","entry":[]}'
+        secret = "test_geheim"
+        sig = "sha256=" + hmac.new(
+            secret.encode("utf-8"), payload, hashlib.sha256
+        ).hexdigest()
+
+        with patch("src.routes.whatsapp.verify_whatsapp_signature",
+                   return_value=True):
+            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":[]}'
+
+        response = self.client.post(
+            "/webhook/whatsapp",
+            data=payload,
+            headers={"X-Hub-Signature-256": "sha256=falsch"},
+            content_type="application/json",
+        )
+        assert response.status_code == 401
+        data = response.get_json()
+        assert "error" in data