hermes
31b1a62cb2
Task 23: Worker-Intake-Pipeline verbinden
- ergebnis-Feld in Submission/Payload
- worker.py extrahiert + worker_loop refactor
- test_worker.py mit VerarbeiteSubmission/WorkerLoop tests
Task 24: Admin-Stats mit echten Daten
- processed_total, uptime, error_rate, VERSION
- admin_health mit Redis-Live-Daten
Task 25: E2E-Integrationstest + worker_refaktorisiert
Quality-Gate-Fixes:
- consumername Collision: 'worker-1' -> f'worker-{id(self)}'
- fakeredis block=0 Timeout fix
- WhatsApp/Telegram webhook payload fixes
- test_telegram/integration/e2e/refaktorisiert
- handler.py recovery nach corrupted Patch
82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
"""Tests für Webhook-Signatur-Prüfung."""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from src.queue.webhook import (
|
|
verify_telegram_signature,
|
|
verify_whatsapp_signature,
|
|
)
|
|
|
|
|
|
class TestWhatsAppSignature:
|
|
"""Tests für WhatsApp-Signatur-Prüfung."""
|
|
|
|
def setup_method(self) -> None:
|
|
"""Setze Test-secret per Environment var."""
|
|
self._old_secret = os.environ.pop("WHATSAPP_APP_SECRET", None)
|
|
os.environ["WHATSAPP_APP_SECRET"] = "test_app_secret_123"
|
|
|
|
def teardown_method(self) -> None:
|
|
"""Environment bereinigen."""
|
|
if self._old_secret:
|
|
os.environ["WHATSAPP_APP_SECRET"] = self._old_secret
|
|
else:
|
|
os.environ.pop("WHATSAPP_APP_SECRET", None)
|
|
|
|
def test_korrekte_whatsapp_signatur(self) -> None:
|
|
"""Korrekte WhatsApp-Signatur → True."""
|
|
payload = b'{"object":"whatsapp","entry":[]}'
|
|
signature = (
|
|
"sha256=" + hmac.new(b"test_app_secret_123", payload, hashlib.sha256).hexdigest()
|
|
)
|
|
|
|
assert verify_whatsapp_signature(payload, signature) is True
|
|
|
|
def test_falsche_whatsapp_signatur(self) -> None:
|
|
"""Falsche WhatsApp-Signatur → False."""
|
|
payload = b'{"object":"whatsapp","entry":[]}'
|
|
signature = "sha256=falschfalschfalschfalschfalschfalsch"
|
|
|
|
assert verify_whatsapp_signature(payload, signature) is False
|
|
|
|
def test_fehlender_app_secret(self) -> None:
|
|
"""Fehlender App-Secret → False (sicherer Fail)."""
|
|
os.environ.pop("WHATSAPP_APP_SECRET", None)
|
|
|
|
assert verify_whatsapp_signature(b"payload", "sha256=xyz") is False
|
|
assert verify_whatsapp_signature(b"payload", "sha256=xyz", app_secret="") is False
|
|
|
|
|
|
class TestTelegramSignature:
|
|
"""Tests für Telegram-Signatur-Prüfung."""
|
|
|
|
def setup_method(self) -> None:
|
|
"""Setze Test-token per Environment var."""
|
|
self._old_token = os.environ.pop("TELEGRAM_BOT_TOKEN", None)
|
|
os.environ["TELEGRAM_BOT_TOKEN"] = "1234567890:***"
|
|
|
|
def teardown_method(self) -> None:
|
|
"""Environment bereinigen."""
|
|
if self._old_token:
|
|
os.environ["TELEGRAM_BOT_TOKEN"] = self._old_token
|
|
else:
|
|
os.environ.pop("TELEGRAM_BOT_TOKEN", None)
|
|
|
|
def test_korrekter_telegram_token(self) -> None:
|
|
"""Korrekter Telegram-Token → True (HMAC stimmt überein)."""
|
|
payload = b"payload"
|
|
secret = "1234567890:***"
|
|
expected = hmac.new(secret.encode("utf-8"), payload, hashlib.sha256).hexdigest()
|
|
assert verify_telegram_signature(payload, expected) is True
|
|
|
|
def test_zu_kurzer_telegram_token(self) -> None:
|
|
"""Ungültiger Telegram-Token (zu kurz) → False."""
|
|
os.environ.pop("TELEGRAM_BOT_TOKEN", None)
|
|
|
|
assert verify_telegram_signature(b"payload", "short") is False
|
|
assert verify_telegram_signature(b"", "") is False
|