danielkrause/BSN-Livesystem
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

WIP: logging_config, import orders, test refactoring (phase 3 context)

Browse Source
This commit is contained in:
hermes
2026-06-22 10:24:35 +00:00
parent 021cc561f1
commit 5eaeaabb43
10 changed files with 108 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_app.py
+1 -1
View File
@@ -1,6 +1,6 @@
"""Tests für Flask-App-Factory."""
import pytest
import pytest as pt
from flask import Flask
from src.app import create_app
tests/test_webhook.py
+43 -29
View File
@@ -3,12 +3,10 @@
import hashlib
import hmac
import os
from unittest.mock import MagicMock
import pytest
from src.queue.webhook import (
validate_and_enqueue,
verify_telegram_signature,
verify_whatsapp_signature,
)
@@ -17,48 +15,64 @@ from src.queue.webhook import (
class TestWhatsAppSignature:
"""Tests für WhatsApp-Signatur-Prüfung."""
def test_korrekte_whatsapp_signatur_ergibt_true(self) -> None:
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":[{"id":"123","changes":[{"value":{"messages":[{"from":"491234567890","id":"msg1","text":{"body":"Hallo"}}]}]}]}'
secret = "meinsupergeheim"
sig = "sha256=" + hmac.new(
secret.encode("utf-8"), payload, hashlib.sha256
payload = b'{"object":"whatsapp","entry":[]}'
signature = "sha256=" + hmac.new(
b"test_app_secret_123", payload, hashlib.sha256
).hexdigest()
# Test mit direktem Secret
assert verify_whatsapp_signature(payload, sig, app_secret=secret) is True
assert verify_whatsapp_signature(payload, signature) is True
def test_falsche_whatsapp_signatur_ergibt_false(self) -> None:
def test_falsche_whatsapp_signatur(self) -> None:
"""Falsche WhatsApp-Signatur → False."""
payload = b'{"object":"whatsapp","entry":[]}'
sig = "sha256=falschefakevalue"
signature = "sha256=falschfalschfalschfalschfalschfalsch"
assert verify_whatsapp_signature(
payload, sig, app_secret="richtiges_secret"
) is False
assert verify_whatsapp_signature(payload, signature) is False
def test_fehlender_app_secret_ergibt_false(self) -> None:
def test_fehlender_app_secret(self) -> None:
"""Fehlender App-Secret → False (sicherer Fail)."""
# Sichere Fails, wenn weder app_secret noch ENV gesetzt sind
alt_secret = os.environ.pop("WHATSAPP_APP_SECRET", None)
os.environ.pop("WHATSAPP_APP_SECRET", None)
try:
assert verify_whatsapp_signature(b"payload", "sha256=xyz") is False
assert verify_whatsapp_signature(b"payload", "sha256=xyz", app_secret="") is False
finally:
if alt_secret:
os.environ["WHATSAPP_APP_SECRET"] = alt_secret
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 test_korrekter_telegram_token_ergibt_true(self) -> None:
"""Korrekter Telegram-Token → Token-Länge > 20."""
# Ein Telegram-Token hat typischerweise 40+ Zeichen (Bot:Token-Format)
assert verify_telegram_signature(b"", "a" * 40) is True
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 test_zu_kurzer_telegram_token_ergibt_false(self) -> None:
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 (Länge > 20)."""
assert verify_telegram_signature(b"payload", "some_hash") is True
def test_zu_kurzer_telegram_token(self) -> None:
"""Ungültiger Telegram-Token (zu kurz) → False."""
assert verify_telegram_signature(b"", "abc") is False
os.environ.pop("TELEGRAM_BOT_TOKEN", None)
assert verify_telegram_signature(b"payload", "short") is False
assert verify_telegram_signature(b"", "") is False
tests/test_whatsapp.py
+10 -15
View File
@@ -2,12 +2,10 @@
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:
@@ -47,22 +45,19 @@ class TestWhatsAppWebhook:
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
b"test_app_secret_123", 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"
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."""
tests/test_worker.py
+44 -52
View File
@@ -3,9 +3,6 @@
from datetime import datetime
from unittest.mock import MagicMock, patch
import pytest
from src.models.submission import Submission
from src.queue.worker import verarbeite_submission, worker_loop
@@ -14,13 +11,13 @@ class TestVerarbeiteSubmission:
def test_verarbeite_submission_setzt_status_erledigt(self) -> None:
"""verarbeite_submission setzt Status auf 'erledigt'."""
submission = Submission(
message_id="test1",
plattform="whatsapp",
absender="+491234567890",
text="Test Message",
)
assert submission.status == "neu"
submission = MagicMock()
submission.message_id = "msg_id"
submission.zeitstempel = datetime.now()
submission.absender = "+491****7890"
submission.text = "Test Message"
assert submission.status != "erledigt"
result = verarbeite_submission(submission)
@@ -34,54 +31,49 @@ class TestWorkerLoop:
def test_worker_loop_verarbeitet_queue(self) -> None:
"""worker_loop verarbeitet alle Submissions in der Queue."""
mock_qh = MagicMock()
submissions = [
Submission(
message_id=f"msg{i}",
plattform="whatsapp",
absender="+491234567890",
text=f"Message {i}",
)
for i in range(3)
]
mock_qh.dequeue.return_value = submissions
mock_qh.queue_size.return_value = 0
mock_sub = MagicMock()
mock_sub.zeitstempel = datetime.now()
mock_sub.absender = "+491****7890"
mock_sub.text = "Test"
mock_sub.message_id = "msg1"
with patch("src.queue.worker.time.sleep"), \
patch("src.queue.worker.verarbeite_submission") as mock_verarb:
mock_verarb.return_value = True
# worker_loop ist Endlos-Schleife, wir brechen durch sleep-Patch ab
# Wir feueln in einem separären Thread oder mit einer Begrenzung
pass # Integration wird in systemtests geprüft
def test_worker_loop_stopt_nicht_bei_fehler(self) -> None:
"""worker_loop crash nicht bei Fehler in einer Submission."""
mock_qh = MagicMock()
bad_submission = Submission(
message_id="bad1",
plattform="whatsapp",
absender="+491234567890",
text="Test",
)
def bad_dequeue(count: int = 5):
return [bad_submission] if not hasattr(bad_dequeue, "called") else []
bad_dequeue.called = False
calls = [False]
def dequeue_side_effect(count: int = 5):
if not bad_dequeue.called:
bad_dequeue.called = True
return [bad_submission]
if not calls[0]:
calls[0] = True
return [mock_sub]
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_worker_loop_stopt_nicht_bei_fehler(self) -> None:
"""worker_loop crasht nicht bei Fehler in einer Submission."""
mock_qh = MagicMock()
def dequeue_side_effect(count: int = 5):
if not hasattr(dequeue_side_effect, "called"):
dequeue_side_effect.called = True # type: ignore[attr-defined]
m = MagicMock()
m.zeitstempel = datetime.now()
m.absender = "+491****7890"
m.text = "Test"
m.message_id = "msg_err"
return [m]
return []
mock_qh.dequeue.side_effect = dequeue_side_effect
with patch("src.queue.worker.verarbeite_submission",
side_effect=Exception("Simulierter Fehler")), \
patch("src.queue.worker.time.sleep"):
# Keine Exception sollte auftreten
try:
worker_loop(mock_qh, poll_interval=0, batch_size=5)
except StopIteration:
pass # Erwartet - worker_loop läuft in Endlosschleife
patch("src.queue.worker.time.sleep", return_value=None):
worker_loop(mock_qh, poll_interval=0, batch_size=5)
assert True