integrity_scanner_fuer_statische_Webseiten/tests/test_alerter.py

"""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

    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)