hermes
138 lines
5.1 KiB
Python
138 lines
5.1 KiB
Python
"""Integration-Tests für WhatsApp/Telegram-Webhook-Flows."""
|
|
|
|
import hashlib
|
|
import hmac
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from flask import Flask
|
|
import pytest
|
|
|
|
from src.app import create_app
|
|
from src.routes.whatsapp import whatsapp_bp
|
|
from src.routes.telegram import telegram_bp
|
|
from src.routes.admin import admin_bp
|
|
from src.queue.worker import verarbeite_submission, worker_loop
|
|
|
|
|
|
@pytest.fixture()
|
|
def flask_app() -> Flask:
|
|
"""Erstellt Flask-App mit allen Blueprints."""
|
|
app = create_app()
|
|
app.queue_handler = MagicMock()
|
|
app.queue_handler.queue_size.return_value = 0
|
|
app.register_blueprint(whatsapp_bp, url_prefix="/")
|
|
app.register_blueprint(telegram_bp, url_prefix="/")
|
|
app.register_blueprint(admin_bp, url_prefix="/")
|
|
return app
|
|
|
|
|
|
class TestWebhookIntegration:
|
|
"""Tests für Webhook-Integration."""
|
|
|
|
def test_whatsapp_post_mit_gueltiger_signatur(self, flask_app: Flask) -> None:
|
|
"""WhatsApp POST mit gültiger Signatur → 200, Submission in Queue."""
|
|
with patch("src.routes.whatsapp.verify_whatsapp_signature", return_value=True):
|
|
response = flask_app.test_client().post(
|
|
"/webhook/whatsapp",
|
|
data=b'{"object":"whatsapp","entry":[]}',
|
|
headers={"X-Hub-Signature-256": "sha256=test"},
|
|
content_type="application/json",
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data["status"] == "accepted"
|
|
flask_app.queue_handler.enqueue.assert_called_once()
|
|
|
|
def test_whatsapp_post_mit_ungueltiger_signatur(self, flask_app: Flask) -> None:
|
|
"""WhatsApp POST mit ungültiger Signatur → 401."""
|
|
with patch("src.routes.whatsapp.verify_whatsapp_signature", return_value=False):
|
|
response = flask_app.test_client().post(
|
|
"/webhook/whatsapp",
|
|
data=b'{"object":"whatsapp","entry":[]}',
|
|
headers={"X-Hub-Signature-256": "sha256=fake"},
|
|
content_type="application/json",
|
|
)
|
|
assert response.status_code == 401
|
|
data = response.get_json()
|
|
assert "error" in data
|
|
|
|
|
|
class TestTelegramIntegration:
|
|
"""Tests für Telegram-Integration."""
|
|
|
|
def test_telegram_post_mit_gueltigem_payload(self, flask_app: Flask) -> None:
|
|
"""Telegram POST mit gültigem Payload → 200."""
|
|
response = flask_app.test_client().post(
|
|
"/webhook/telegram",
|
|
data={"message": {"message_id": "123", "chat": {"id": "456"}, "text": "Hallo"}},
|
|
content_type="application/json",
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data["status"] == "accepted"
|
|
flask_app.queue_handler.enqueue.assert_called_once()
|
|
|
|
|
|
class TestWorkerIntegration:
|
|
"""Tests für Worker-Integration."""
|
|
|
|
def test_vollstaendiger_flow_whatsapp(self, flask_app: Flask) -> None:
|
|
"""Vollständiger Flow: WhatsApp POST → Worker verarbeitet → Status 'erledigt'."""
|
|
mock_qh = flask_app.queue_handler
|
|
mock_submission = MagicMock()
|
|
mock_submission.message_id = "msg001"
|
|
mock_submission.zeitstempel = "2024-01-01T00:00:00"
|
|
mock_submission.absender = "+491****7890"
|
|
mock_submission.text = "suche Catan"
|
|
mock_submission.status = "neu"
|
|
mock_submission.enqueue.return_value = "stream_id"
|
|
|
|
# Simuliere: Submit wurde enqueued, dann vom Worker geholt
|
|
call_count = [0]
|
|
|
|
def dequeue_side_effect(count: int = 5):
|
|
call_count[0] += 1
|
|
if call_count[0] <= 1:
|
|
return [mock_submission]
|
|
return []
|
|
|
|
mock_qh.dequeue.side_effect = dequeue_side_effect
|
|
|
|
with (
|
|
patch("src.queue.worker.verarbeite_submission", return_value=True) as mock_verarb,
|
|
patch("src.queue.worker.time.sleep", return_value=None),
|
|
):
|
|
worker_loop(mock_qh, poll_interval=0, batch_size=5)
|
|
|
|
assert mock_verarb.call_count == 1
|
|
assert mock_qh.mark_done.call_count == 1
|
|
|
|
def test_vollstaendiger_flow_telegram(self, flask_app: Flask) -> None:
|
|
"""Vollständiger Flow: Telegram POST → Worker verarbeitet → Status 'erledigt'."""
|
|
mock_qh = flask_app.queue_handler
|
|
mock_submission = MagicMock()
|
|
mock_submission.message_id = "msg002"
|
|
mock_submission.zeitstempel = "2024-01-01T00:00:00"
|
|
mock_submission.absender = "@user123"
|
|
mock_submission.text = "hilfe"
|
|
mock_submission.status = "neu"
|
|
|
|
call_count = [0]
|
|
|
|
def dequeue_side_effect(count: int = 5):
|
|
call_count[0] += 1
|
|
if call_count[0] <= 1:
|
|
return [mock_submission]
|
|
return []
|
|
|
|
mock_qh.dequeue.side_effect = dequeue_side_effect
|
|
|
|
with (
|
|
patch("src.queue.worker.verarbeite_submission", return_value=True) as mock_verarb,
|
|
patch("src.queue.worker.time.sleep", return_value=None),
|
|
):
|
|
worker_loop(mock_qh, poll_interval=0, batch_size=5)
|
|
|
|
assert mock_verarb.call_count == 1
|
|
assert mock_qh.mark_done.call_count == 1
|