test(#134): monitoring.py Coverage 83.2% → 99.3%

- TestSearchAdapterFallbackLogging: erster Query-Versuch failt mit
  Debug-Log, dritter klappt
- TestDailyScanDbUpsertFailure: erster upsert_monitoring_scan crasht,
  zweiter klappt → der Rest des Protokolls wird nicht blockiert,
  ERROR-Log ist da
- TestSendMonitoringDigest:
  - mail_sent=True bei erfolgreichem send_mail
  - mail_sent=False bei SMTP-Fehler, aber kein Crash

Verbleibend: Line 122 (return [] nach drei Fallback-Misses ohne
Exception — schwer ohne Adapter-Mock zu provozieren).

Total Coverage: 49.5% → 49.9%, 714 → 718 Tests.
This commit is contained in:
Dotty Dotter 2026-04-28 11:01:19 +02:00
parent e69ca1c29d
commit 999926b5f3

View File

@ -351,3 +351,156 @@ class TestRenderPlain:
] ]
text = _render_plain(self._make_result(), docs) text = _render_plain(self._make_result(), docs)
assert "weitere" not in text assert "weitere" not in text
# ─── Coverage-Backfill (#134) ────────────────────────────────────────────────
class TestSearchAdapterFallbackLogging:
"""Erste/zweite Query schlagen fehl → Debug-Log, dritter klappt."""
def test_fallback_logs_intermediate_failures(self, caplog):
import asyncio
import logging
from app.monitoring import _search_adapter
class FakeAdapter:
calls = 0
async def search(self, q, limit):
FakeAdapter.calls += 1
if FakeAdapter.calls < 3:
raise RuntimeError(f"transient {FakeAdapter.calls}")
return ["ok"]
with caplog.at_level(logging.DEBUG, logger="app.monitoring"):
result = asyncio.run(_search_adapter(FakeAdapter(), "BX"))
assert result == ["ok"]
class TestDailyScanDbUpsertFailure:
"""Wenn upsert_monitoring_scan fuer einzelne Drucksache crasht,
wird der Rest weiter verarbeitet (Line 191-192).
Adapter werden aus app.parlamente.ADAPTERS importiert also
monkey-patchen wir dort.
"""
def test_upsert_exception_logged_and_skipped(self, monkeypatch, caplog):
import asyncio
import logging
from types import SimpleNamespace
from app import monitoring as mon
from app.bundeslaender import Bundesland
import app.parlamente as parl_mod
import app.database as db_mod
# Adapter mit zwei Drucksachen
class FakeAdapter:
async def search(self, q, limit):
return [
SimpleNamespace(bundesland="BX", drucksache="1/1",
title="A1", datum="2026-04-01",
typ="Antrag", typ_normiert="antrag",
fraktionen=["CDU"], link="https://x"),
SimpleNamespace(bundesland="BX", drucksache="1/2",
title="A2", datum="2026-04-02",
typ="Antrag", typ_normiert="antrag",
fraktionen=["SPD"], link="https://y"),
]
# Erster upsert wirft, zweiter klappt
call_count = {"n": 0}
async def fake_upsert(**kw):
call_count["n"] += 1
if call_count["n"] == 1:
raise RuntimeError("DB-Lock")
return True
async def fake_summary(**kw):
return None
# Fake-BL fuer aktive_bundeslaender
fake_bl = Bundesland(
code="BX", name="Test-BL", parlament_name="Test", wahlperiode=1,
wahlperiode_start="2024-01-01", naechste_wahl=None,
regierungsfraktionen=[], landtagsfraktionen=[],
doku_system="Test", doku_base_url="http://example.com",
drucksache_format="1/1234", dokukratie_scraper=None, aktiv=True,
)
monkeypatch.setattr(mon, "aktive_bundeslaender", lambda: [fake_bl])
# Adapter-Dict im parlamente-Modul (von dem mon importiert)
monkeypatch.setitem(parl_mod.ADAPTERS, "BX", FakeAdapter())
monkeypatch.setattr(db_mod, "upsert_monitoring_scan", fake_upsert)
monkeypatch.setattr(db_mod, "upsert_monitoring_summary", fake_summary)
with caplog.at_level(logging.ERROR, logger="app.monitoring"):
result = asyncio.run(mon.daily_scan(limit=10))
bx_results = [r for r in result.results if r.bundesland == "BX"]
assert len(bx_results) == 1
# Erster crashte → new_count=1 (zweiter klappte)
assert bx_results[0].new_count == 1
assert call_count["n"] == 2
assert any("DB-UPSERT fehlgeschlagen" in r.message for r in caplog.records)
class TestSendMonitoringDigest:
"""run_monitoring_digest rendert Template, ruft send_mail."""
def test_mail_sent_returns_true(self, monkeypatch, tmp_path):
import asyncio
from app import monitoring as mon
async def fake_scan(**kw):
return mon.DailyScanResult(
scan_date="2026-04-28",
results=[], new_total=0, total_seen=0,
estimated_cost_eur=0.0, errors=[],
)
async def fake_get_new_today(scan_date):
return []
async def fake_send_mail(to, subj, text, html):
return None
monkeypatch.setattr(mon, "daily_scan", fake_scan)
# Importer-Patches innerhalb der Funktion sind tricky — wir patchen
# stattdessen die Module-Funktionen direkt
import app.mail
import app.database
monkeypatch.setattr(app.mail, "send_mail", fake_send_mail)
monkeypatch.setattr(app.database, "get_monitoring_new_today",
fake_get_new_today)
result = asyncio.run(mon.run_monitoring_digest("admin@test"))
assert result["mail_sent"] is True
assert result["scan_date"] == "2026-04-28"
def test_mail_failure_returns_false_but_not_raises(self, monkeypatch):
import asyncio
from app import monitoring as mon
async def fake_scan(**kw):
return mon.DailyScanResult(
scan_date="2026-04-28",
results=[], new_total=0, total_seen=0,
estimated_cost_eur=0.0, errors=[],
)
async def fake_get_new_today(scan_date):
return []
async def failing_send_mail(to, subj, text, html):
raise ConnectionError("SMTP down")
monkeypatch.setattr(mon, "daily_scan", fake_scan)
import app.mail
import app.database
monkeypatch.setattr(app.mail, "send_mail", failing_send_mail)
monkeypatch.setattr(app.database, "get_monitoring_new_today",
fake_get_new_today)
result = asyncio.run(mon.run_monitoring_digest("admin@test"))
assert result["mail_sent"] is False