Hermes Agent
97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
"""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
|