test: add test_alerter.py — 25 tests for send_alert, email and webhook dispatch
Covers: level filtering, email enabled/disabled, TLS, login with/without password, no recipients, SMTP exception handling, webhook payload, webhook exception handling. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
beb7f274a3
commit
fe61b18906
1 changed files with 251 additions and 0 deletions
251
tests/test_alerter.py
Normal file
251
tests/test_alerter.py
Normal file
|
|
@ -0,0 +1,251 @@
|
|||
"""Tests for scanner/alerter.py — alert dispatch (e-mail + webhook)."""
|
||||
import os
|
||||
from unittest.mock import MagicMock, call, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from scanner.alerter import _make_subject, _format_body, send_alert
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _report(level: str = "yellow", target: str = "https://example.de") -> dict:
|
||||
return {
|
||||
"target": target,
|
||||
"assessment": {
|
||||
"level": level,
|
||||
"score": 30 if level == "yellow" else 80,
|
||||
"reasons": ["[TEST] Testbefund"],
|
||||
"exit_code": 1 if level == "yellow" else 2,
|
||||
},
|
||||
"diff": {
|
||||
"new_external_domains": [],
|
||||
"new_internal_urls": [],
|
||||
"missing_internal_urls": [],
|
||||
"page_diffs": [],
|
||||
"new_broken_urls": [],
|
||||
"known_broken_urls": [],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _cfg(
|
||||
enabled: bool = True,
|
||||
min_level: str = "yellow",
|
||||
recipients: list | None = None,
|
||||
pw_env: str = "TEST_SMTP_PW",
|
||||
webhook_enabled: bool = False,
|
||||
webhook_url: str = "",
|
||||
) -> dict:
|
||||
return {
|
||||
"alerting": {
|
||||
"min_level": min_level,
|
||||
"email": {
|
||||
"enabled": enabled,
|
||||
"smtp_host": "mail.example.de",
|
||||
"smtp_port": 587,
|
||||
"smtp_tls": True,
|
||||
"smtp_user": "scanner",
|
||||
"smtp_password_env": pw_env,
|
||||
"from": "scanner@example.de",
|
||||
"to": recipients if recipients is not None else ["admin@example.de"],
|
||||
},
|
||||
"webhook": {
|
||||
"enabled": webhook_enabled,
|
||||
"url": webhook_url,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _make_subject
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestMakeSubject:
|
||||
def test_yellow_subject_contains_target(self):
|
||||
subj = _make_subject(_report("yellow", "https://meine-seite.de"))
|
||||
assert "meine-seite.de" in subj
|
||||
|
||||
def test_red_subject_contains_aufmerksamkeit(self):
|
||||
subj = _make_subject(_report("red"))
|
||||
assert "Aufmerksamkeit" in subj
|
||||
|
||||
def test_green_subject_contains_ordnung(self):
|
||||
subj = _make_subject(_report("green"))
|
||||
assert "Ordnung" in subj
|
||||
|
||||
def test_each_level_has_emoji(self):
|
||||
for level, emoji in [("green", "🟢"), ("yellow", "🟡"), ("red", "🔴")]:
|
||||
assert emoji in _make_subject(_report(level))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _format_body
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestFormatBody:
|
||||
def test_body_contains_target(self):
|
||||
body = _format_body(_report("yellow", "https://meine-seite.de"))
|
||||
assert "meine-seite.de" in body
|
||||
|
||||
def test_body_contains_level(self):
|
||||
body = _format_body(_report("red"))
|
||||
assert "RED" in body
|
||||
|
||||
def test_body_contains_reasons(self):
|
||||
body = _format_body(_report("yellow"))
|
||||
assert "[TEST] Testbefund" in body
|
||||
|
||||
def test_body_contains_workflow_hint(self):
|
||||
body = _format_body(_report("yellow"))
|
||||
assert "python -m scanner" in body
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# send_alert — level filtering
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSendAlertLevelFilter:
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_green_below_yellow_min_suppressed(self, mock_smtp):
|
||||
send_alert(_report("green"), _cfg(min_level="yellow"))
|
||||
mock_smtp.assert_not_called()
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_yellow_meets_yellow_min_sent(self, mock_smtp, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "secret")
|
||||
send_alert(_report("yellow"), _cfg(min_level="yellow"))
|
||||
mock_smtp.assert_called_once()
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_red_meets_yellow_min_sent(self, mock_smtp, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "secret")
|
||||
send_alert(_report("red"), _cfg(min_level="yellow"))
|
||||
mock_smtp.assert_called_once()
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_yellow_below_red_min_suppressed(self, mock_smtp):
|
||||
send_alert(_report("yellow"), _cfg(min_level="red"))
|
||||
mock_smtp.assert_not_called()
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_red_meets_red_min_sent(self, mock_smtp, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "secret")
|
||||
send_alert(_report("red"), _cfg(min_level="red"))
|
||||
mock_smtp.assert_called_once()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# send_alert — e-mail enabled/disabled
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSendAlertEmail:
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_email_disabled_no_smtp(self, mock_smtp):
|
||||
send_alert(_report("yellow"), _cfg(enabled=False))
|
||||
mock_smtp.assert_not_called()
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_no_recipients_no_send(self, mock_smtp, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "secret")
|
||||
send_alert(_report("yellow"), _cfg(recipients=[]))
|
||||
mock_smtp.assert_not_called()
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_starttls_called_when_tls_enabled(self, mock_smtp, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "secret")
|
||||
server = MagicMock()
|
||||
mock_smtp.return_value.__enter__ = MagicMock(return_value=server)
|
||||
mock_smtp.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
send_alert(_report("yellow"), _cfg())
|
||||
server.starttls.assert_called_once()
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_starttls_not_called_when_tls_disabled(self, mock_smtp, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "secret")
|
||||
server = MagicMock()
|
||||
mock_smtp.return_value.__enter__ = MagicMock(return_value=server)
|
||||
mock_smtp.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
cfg = _cfg()
|
||||
cfg["alerting"]["email"]["smtp_tls"] = False
|
||||
send_alert(_report("yellow"), cfg)
|
||||
server.starttls.assert_not_called()
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_login_called_with_user_and_password(self, mock_smtp, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "geheim")
|
||||
server = MagicMock()
|
||||
mock_smtp.return_value.__enter__ = MagicMock(return_value=server)
|
||||
mock_smtp.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
send_alert(_report("yellow"), _cfg())
|
||||
server.login.assert_called_once_with("scanner", "geheim")
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_login_skipped_when_no_password(self, mock_smtp, monkeypatch):
|
||||
monkeypatch.delenv("TEST_SMTP_PW", raising=False)
|
||||
server = MagicMock()
|
||||
mock_smtp.return_value.__enter__ = MagicMock(return_value=server)
|
||||
mock_smtp.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
send_alert(_report("yellow"), _cfg())
|
||||
server.login.assert_not_called()
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_send_message_called(self, mock_smtp, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "secret")
|
||||
server = MagicMock()
|
||||
mock_smtp.return_value.__enter__ = MagicMock(return_value=server)
|
||||
mock_smtp.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
send_alert(_report("yellow"), _cfg())
|
||||
server.send_message.assert_called_once()
|
||||
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_smtp_exception_does_not_raise(self, mock_smtp, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "secret")
|
||||
mock_smtp.side_effect = ConnectionRefusedError("Connection refused")
|
||||
# Must not propagate
|
||||
send_alert(_report("yellow"), _cfg())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# send_alert — webhook
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSendAlertWebhook:
|
||||
@patch("scanner.alerter.requests.post")
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_webhook_called_when_enabled(self, mock_smtp, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "secret")
|
||||
cfg = _cfg(webhook_enabled=True, webhook_url="https://hooks.example.de/abc")
|
||||
send_alert(_report("yellow"), cfg)
|
||||
mock_post.assert_called_once()
|
||||
assert mock_post.call_args[0][0] == "https://hooks.example.de/abc"
|
||||
|
||||
@patch("scanner.alerter.requests.post")
|
||||
def test_webhook_not_called_when_disabled(self, mock_post):
|
||||
send_alert(_report("yellow"), _cfg(enabled=False, webhook_enabled=False))
|
||||
mock_post.assert_not_called()
|
||||
|
||||
@patch("scanner.alerter.requests.post")
|
||||
@patch("scanner.alerter.smtplib.SMTP")
|
||||
def test_webhook_payload_contains_level(self, mock_smtp, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("TEST_SMTP_PW", "secret")
|
||||
cfg = _cfg(webhook_enabled=True, webhook_url="https://hooks.example.de/abc")
|
||||
send_alert(_report("red"), cfg)
|
||||
payload = mock_post.call_args[1]["json"]
|
||||
assert payload["level"] == "red"
|
||||
|
||||
@patch("scanner.alerter.requests.post")
|
||||
def test_webhook_exception_does_not_raise(self, mock_post):
|
||||
mock_post.side_effect = Exception("Network error")
|
||||
cfg = _cfg(enabled=False, webhook_enabled=True, webhook_url="https://hooks.example.de/x")
|
||||
# green is below min_level yellow → suppressed anyway, but let's test enabled+green=red path
|
||||
cfg["alerting"]["min_level"] = "green"
|
||||
send_alert(_report("green"), cfg)
|
||||
Loading…
Add table
Add a link
Reference in a new issue