"""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 "[ALARM]" in body # rotes Level-Badge im technischen Teil assert "🔴" in body # Ampel-Symbol im Klartext-Teil def test_body_equals_report_renderer(self): """E-Mail-Body muss exakt der Dateibericht sein (eine Quelle der Wahrheit).""" from scanner.report import render_markdown rep = _report("yellow") assert _format_body(rep) == render_markdown(rep) def test_commands_include_target_url(self): """Alle Scanner-Kommandos in der Mail enthalten das Target (direkt ausführbar).""" body = _format_body(_report("yellow", "https://meine-seite.de")) assert "python -m scanner https://meine-seite.de approve --all" in body def test_score_reflects_findings_not_just_zero(self): """Bei YELLOW darf der angezeigte Score nicht 0 sein, wenn Gründe vorliegen.""" rep = _report("yellow") rep["assessment"]["score"] = 40 # aggregierter Gesamt-Score (z. B. aus KI) assert "Score:** 40" in _format_body(rep) 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 def test_ai_finding_includes_url_and_reason(self): """Beanstandete Dateien müssen mit URL und Begründung im Body stehen.""" rep = _report("yellow") rep["ai_result"] = { "checked": 1, "cache_hits": 0, "api_calls": 1, "skipped": None, "findings": [{ "kind": "image", "url": "https://x.de/galerie", "asset_url": "https://x.de/img/bad.jpg", "fingerprint": "a1b2c3d4", "category": "pornography", "severity": "high", "confidence": 0.95, "explanation": "Explizite Darstellung, passt nicht zum Thema.", }], } body = _format_body(rep) assert "https://x.de/img/bad.jpg" in body # URL der Datei assert "Explizite Darstellung" in body # Begründung assert "ai-dismiss --hash a1b2c3d4" in body # Quittier-Hinweis # --------------------------------------------------------------------------- # 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)