Hermes Agent
112 lines
3.1 KiB
Python
112 lines
3.1 KiB
Python
"""Tests für den Telegram-Webhook Blueprint."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fakeredis
|
|
from flask import Flask
|
|
from flask.testing import FlaskClient
|
|
|
|
from src.app import create_app
|
|
from src.queue.handler import QueueHandler
|
|
from src.routes.telegram import telegram
|
|
|
|
|
|
def _setup_app_with_fakeredis() -> Flask:
|
|
"""Erstellt eine Flask-App mit fakeredis-QueueHandler und registriert den Telegram-Blueprint."""
|
|
app = create_app()
|
|
|
|
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(telegram)
|
|
|
|
return app
|
|
|
|
|
|
def test_valid_telegram_payload_returns_200() -> None:
|
|
"""POST /webhook/telegram mit gültigem Payload → 200."""
|
|
app = _setup_app_with_fakeredis()
|
|
client: FlaskClient = app.test_client()
|
|
|
|
payload = {
|
|
"update_id": 123456789,
|
|
"message": {
|
|
"message_id": 100,
|
|
"from": {
|
|
"id": 987654321,
|
|
"first_name": "TestUser",
|
|
},
|
|
"chat": {
|
|
"id": 987654321,
|
|
"first_name": "TestUser",
|
|
"type": "private",
|
|
},
|
|
"date": 1719000000,
|
|
"text": "Hallo BSN-Bot!",
|
|
},
|
|
}
|
|
|
|
response = client.post("/webhook/telegram", json=payload)
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data is not None
|
|
assert data["status"] == "ok"
|
|
|
|
|
|
def test_missing_json_returns_400() -> None:
|
|
"""POST /webhook/telegram ohne JSON-Body → 400."""
|
|
app = _setup_app_with_fakeredis()
|
|
client: FlaskClient = app.test_client()
|
|
|
|
response = client.post("/webhook/telegram", data="not json", content_type="text/plain")
|
|
assert response.status_code == 400
|
|
data = response.get_json()
|
|
assert data is not None
|
|
assert "error" in data
|
|
|
|
|
|
def test_missing_message_id_returns_400() -> None:
|
|
"""POST /webhook/telegram ohne message_id → 400."""
|
|
app = _setup_app_with_fakeredis()
|
|
client: FlaskClient = app.test_client()
|
|
|
|
payload = {
|
|
"message": {
|
|
"chat": {"id": 123},
|
|
"text": "Hallo",
|
|
}
|
|
}
|
|
|
|
response = client.post("/webhook/telegram", json=payload)
|
|
assert response.status_code == 400
|
|
data = response.get_json()
|
|
assert data is not None
|
|
assert "message_id" in data["error"]
|
|
|
|
|
|
def test_empty_text_returns_400() -> None:
|
|
"""POST /webhook/telegram mit leerem text → 400 (min_length=1 ValidationError)."""
|
|
app = _setup_app_with_fakeredis()
|
|
client: FlaskClient = app.test_client()
|
|
|
|
payload = {
|
|
"message": {
|
|
"message_id": 200,
|
|
"chat": {"id": 456},
|
|
"text": "",
|
|
}
|
|
}
|
|
|
|
response = client.post("/webhook/telegram", json=payload)
|
|
assert response.status_code == 400
|
|
data = response.get_json()
|
|
assert data is not None
|
|
assert "error" in data
|