dschlueter/integrity_scanner_fuer_statische_Webseiten
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity
integrity_scanner_fuer_stat.../tests/test_alerter.py
Dieter Schlüter 248788e282 fix: M1 exclude_paths '/' jamulix, M2 seed-should_crawl filter, M3 alert dedup 
- M1: Removed '/' from jamulix.de exclude_paths (blockaded ALL paths via startswith)
- M2: Seed URLs now pass through should_crawl() in crawler.py
- M3: Alert deduplication via last_alert.json (hash of reasons+level+score)
- Added 11 tests for alert deduplication (277 total, all green)
2026-06-14 21:12:36 +02:00

390 lines
15 KiB
Python
Raw Permalink Blame History

"""Tests for scanner/alerter.py — alert dispatch (e-mail + webhook)."""
import json
import os
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import MagicMock, call, patch
import pytest
from scanner.alerter import (
_compute_alert_key,
_load_last_alert,
_save_last_alert,
_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 = "",
data_dir: str | None = None,
) -> dict:
cfg: dict = {
"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,
},
}
}
if data_dir is not None:
cfg["data_dir"] = data_dir
return cfg
# ---------------------------------------------------------------------------
# _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)
# ---------------------------------------------------------------------------
# Deduplication
# ---------------------------------------------------------------------------
class TestComputeAlertKey:
def test_same_reasons_same_key(self):
r1 = _report("red", "https://a.de")
r2 = _report("red", "https://b.de") # anderer Target, gleiche reasons/level/score
assert _compute_alert_key(r1) == _compute_alert_key(r2)
def test_different_score_different_key(self):
r1 = _report("yellow") # score=30
r2 = _report("red") # score=80
assert _compute_alert_key(r1) != _compute_alert_key(r2)
def test_different_reasons_different_key(self):
r1 = _report("yellow")
r1["assessment"]["reasons"] = ["Grund A"]
r2 = _report("yellow")
r2["assessment"]["reasons"] = ["Grund B"]
assert _compute_alert_key(r1) != _compute_alert_key(r2)
def test_reasons_order_independent(self):
r1 = _report("yellow")
r1["assessment"]["reasons"] = ["B", "A", "C"]
r2 = _report("yellow")
r2["assessment"]["reasons"] = ["C", "A", "B"] # gleiche, andere Reihenfolge
assert _compute_alert_key(r1) == _compute_alert_key(r2)
class TestLoadSaveLastAlert:
def test_load_missing_file_returns_empty(self):
with TemporaryDirectory() as tmp:
state = _load_last_alert(Path(tmp))
assert state == {}
def test_load_corrupt_json_returns_empty(self):
with TemporaryDirectory() as tmp:
(Path(tmp) / "last_alert.json").write_text("{invalid")
state = _load_last_alert(Path(tmp))
assert state == {}
def test_save_and_load_roundtrip(self):
with TemporaryDirectory() as tmp:
d = Path(tmp)
_save_last_alert(d, "abc123")
state = _load_last_alert(d)
assert state["key"] == "abc123"
assert "sent_at" in state
class TestSendAlertDedup:
@patch("scanner.alerter._send_email")
def test_first_alert_sent(self, mock_email):
with TemporaryDirectory() as tmp:
cfg = _cfg(data_dir=tmp)
send_alert(_report("yellow"), cfg)
mock_email.assert_called_once()
@patch("scanner.alerter._send_email")
def test_identical_second_alert_suppressed(self, mock_email):
with TemporaryDirectory() as tmp:
cfg = _cfg(data_dir=tmp)
report = _report("yellow")
send_alert(report, cfg)
send_alert(report, cfg) # gleiche reasons → unterdrückt
assert mock_email.call_count == 1
@patch("scanner.alerter._send_email")
def test_changed_reasons_triggers_new_alert(self, mock_email):
with TemporaryDirectory() as tmp:
cfg = _cfg(data_dir=tmp)
r1 = _report("yellow")
r1["assessment"]["reasons"] = ["Grund A"]
send_alert(r1, cfg)
r2 = _report("yellow")
r2["assessment"]["reasons"] = ["Grund B"] # anderer Grund
send_alert(r2, cfg)
assert mock_email.call_count == 2
@patch("scanner.alerter._send_email")
def test_no_data_dir_sends_every_time(self, mock_email):
cfg = _cfg(data_dir=None) # kein data_dir → keine Dedup
send_alert(_report("yellow"), cfg)
send_alert(_report("yellow"), cfg)
assert mock_email.call_count == 2
Reference in a new issue View git blame Copy permalink