b13b46a444
- app/ingest_votes.py 39.2% → 100%
- TestDownloadPdf: schreibt Bytes, propagiert HTTP-Fehler
- TestCli: --supported, kein-arg-error, fehlender PDF-Pfad,
pdf-Pfad-Run, --url-Download-Pfad, exit-Code 2 bei null Resultaten,
Errors-Liste im Output
- DB-Error-Collection in ingest_pdf
- app/wahlprogramme.py 90.7% → 100%
- TestLoadWahlprogrammText: paged-Datei, Normal-Datei-Fallback,
fehlende Datei
- TestSearchWahlprogramm: leere Returns
- TestFindRelevantQuotes: ValueError bei unbekanntem BL
- TestFormatQuoteForPrompt: leeres Dict
- app/abgeordnetenwatch.py 95.2% → 97.6%
- test_rp_pattern_nr_wp_swap: '/538-18.pdf' → '18/538'
- test_sn_pattern_dok_nr_leg_per_swap: 'dok_nr=2150&leg_per=8' → '8/2150'
Total: 47.59% → 48.69%, 666 → 686 Tests, 0 Failures.
421 lines
16 KiB
Python
421 lines
16 KiB
Python
"""Tests für app/abgeordnetenwatch.py und die DB-Funktionen (#106 Phase 1)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import sys
|
||
import types
|
||
|
||
import pytest
|
||
|
||
|
||
# ─── aiosqlite-Stub-Schutz (analog test_database.py) ────────────────────────
|
||
_aio = sys.modules.get("aiosqlite")
|
||
if _aio is not None and not hasattr(_aio, "connect"):
|
||
del sys.modules["aiosqlite"]
|
||
|
||
import aiosqlite as _real_aiosqlite # noqa: E402
|
||
import importlib as _importlib
|
||
|
||
for _mod in ("app.database", "app.abgeordnetenwatch"):
|
||
if _mod in sys.modules:
|
||
_db_mod = sys.modules[_mod]
|
||
if _mod == "app.database" and not hasattr(
|
||
getattr(_db_mod, "aiosqlite", None), "connect"
|
||
):
|
||
del sys.modules[_mod]
|
||
_importlib.import_module(_mod)
|
||
else:
|
||
_importlib.import_module(_mod)
|
||
|
||
|
||
def run(coro):
|
||
return asyncio.get_event_loop().run_until_complete(coro)
|
||
|
||
|
||
# ─── Fixtures ────────────────────────────────────────────────────────────────
|
||
|
||
@pytest.fixture()
|
||
def db_path(tmp_path, monkeypatch):
|
||
path = tmp_path / "test_aw.db"
|
||
from app.config import settings
|
||
monkeypatch.setattr(settings, "db_path", str(path))
|
||
return str(path)
|
||
|
||
|
||
@pytest.fixture()
|
||
def initialized_db(db_path):
|
||
from app import database
|
||
run(database.init_db())
|
||
return db_path
|
||
|
||
|
||
# ─── extract_drucksache_from_intro ───────────────────────────────────────────
|
||
|
||
class TestExtractDrucksache:
|
||
def test_simple_match(self):
|
||
from app.abgeordnetenwatch import extract_drucksache_from_intro
|
||
html = "<p>Beratung des Antrags 18/1234 der Fraktion SPD.</p>"
|
||
assert extract_drucksache_from_intro(html) == "18/1234"
|
||
|
||
def test_match_in_link(self):
|
||
from app.abgeordnetenwatch import extract_drucksache_from_intro
|
||
html = '<a href="...">Drucksache 7/98765</a>'
|
||
assert extract_drucksache_from_intro(html) == "7/98765"
|
||
|
||
def test_first_match_wins(self):
|
||
from app.abgeordnetenwatch import extract_drucksache_from_intro
|
||
html = "Antrag 17/100 und 18/200"
|
||
assert extract_drucksache_from_intro(html) == "17/100"
|
||
|
||
def test_no_match_returns_none(self):
|
||
from app.abgeordnetenwatch import extract_drucksache_from_intro
|
||
html = "<p>Kein Bezug auf eine Drucksache hier.</p>"
|
||
assert extract_drucksache_from_intro(html) is None
|
||
|
||
def test_empty_string_returns_none(self):
|
||
from app.abgeordnetenwatch import extract_drucksache_from_intro
|
||
assert extract_drucksache_from_intro("") is None
|
||
|
||
def test_none_input_returns_none(self):
|
||
from app.abgeordnetenwatch import extract_drucksache_from_intro
|
||
assert extract_drucksache_from_intro(None) is None
|
||
|
||
def test_too_short_sequence_not_matched(self):
|
||
"""Zwei Ziffern reichen nicht aus (min. 3 nach Slash)."""
|
||
from app.abgeordnetenwatch import extract_drucksache_from_intro
|
||
html = "Seite 3/12 — nicht relevant"
|
||
assert extract_drucksache_from_intro(html) is None
|
||
|
||
def test_rp_pattern_nr_wp_swap(self):
|
||
"""RP-URL '/538-18.pdf' → drucksache-Format 'wp/nr' = '18/538'.
|
||
Wir vermeiden im HTML jegliche 'wp/nr'-Notation, sonst greift der
|
||
generische 'Drucksache (\\d+)/(\\d+)'-Match zuerst."""
|
||
from app.abgeordnetenwatch import extract_drucksache_from_intro
|
||
html = '<a href="https://landtag.rlp.de/dokumente/538-18.pdf">Antrag</a>'
|
||
result = extract_drucksache_from_intro(html)
|
||
assert result == "18/538"
|
||
|
||
def test_sn_pattern_dok_nr_leg_per_swap(self):
|
||
"""SN-URL 'dok_nr=2150&...&leg_per=8' → '8/2150'."""
|
||
from app.abgeordnetenwatch import extract_drucksache_from_intro
|
||
html = '<a href="/cgi-bin/foo?dok_nr=2150&extra=x&leg_per=8">DS</a>'
|
||
assert extract_drucksache_from_intro(html) == "8/2150"
|
||
|
||
def test_two_digit_wp_number(self):
|
||
from app.abgeordnetenwatch import extract_drucksache_from_intro
|
||
html = "Bezug: 19/12345"
|
||
assert extract_drucksache_from_intro(html) == "19/12345"
|
||
|
||
|
||
# ─── PARLIAMENT_ID-Mapping ────────────────────────────────────────────────────
|
||
|
||
class TestParliamentIdMapping:
|
||
def test_bt_maps_to_5(self):
|
||
from app.abgeordnetenwatch import PARLIAMENT_ID
|
||
assert PARLIAMENT_ID["BT"] == 5
|
||
|
||
def test_bund_alias_maps_to_5(self):
|
||
from app.abgeordnetenwatch import PARLIAMENT_ID
|
||
assert PARLIAMENT_ID["BUND"] == 5
|
||
|
||
def test_nrw_maps_to_4(self):
|
||
from app.abgeordnetenwatch import PARLIAMENT_ID
|
||
assert PARLIAMENT_ID["NRW"] == 4
|
||
|
||
def test_all_16_bundeslaender_plus_bt_present(self):
|
||
from app.abgeordnetenwatch import PARLIAMENT_ID
|
||
expected_codes = {
|
||
"BT", "NRW", "BE", "HH", "BW", "RP", "LSA", "MV",
|
||
"HB", "HE", "NI", "BY", "SL", "TH", "BB", "SN", "SH",
|
||
}
|
||
assert expected_codes <= set(PARLIAMENT_ID.keys())
|
||
|
||
|
||
# ─── fetch_polls — Stub-Test via httpx mock ───────────────────────────────────
|
||
|
||
class TestFetchPollsStub:
|
||
def test_fetch_polls_parses_response(self, monkeypatch):
|
||
"""Stub-Test: httpx.AsyncClient.get gibt eine Fake-API-Antwort zurück."""
|
||
import httpx
|
||
from app import abgeordnetenwatch as aw_mod
|
||
|
||
fake_polls = [
|
||
{
|
||
"id": 42,
|
||
"label": "Testabstimmung 1",
|
||
"field_poll_date": "2026-04-01",
|
||
"field_accepted": True,
|
||
"field_topics": [{"label": "Klimaschutz"}],
|
||
"field_intro": "<p>Antrag 18/999 der Fraktion GRÜNE</p>",
|
||
"field_legislature": {"label": "18. Wahlperiode"},
|
||
}
|
||
]
|
||
|
||
class FakeResponse:
|
||
def raise_for_status(self): pass
|
||
def json(self): return {"data": fake_polls}
|
||
|
||
class FakeClient:
|
||
async def __aenter__(self): return self
|
||
async def __aexit__(self, *a): pass
|
||
async def get(self, url, params=None): return FakeResponse()
|
||
|
||
monkeypatch.setattr(httpx, "AsyncClient", lambda **kw: FakeClient())
|
||
|
||
polls = run(aw_mod.fetch_polls("NRW", limit=10))
|
||
assert len(polls) == 1
|
||
assert polls[0]["id"] == 42
|
||
assert polls[0]["drucksache"] == "18/999"
|
||
assert polls[0]["field_accepted"] is True
|
||
|
||
def test_fetch_polls_unknown_bundesland_raises(self):
|
||
from app.abgeordnetenwatch import fetch_polls
|
||
with pytest.raises(ValueError, match="Unbekannter BL-Code"):
|
||
run(fetch_polls("XX"))
|
||
|
||
def test_fetch_votes_parses_response(self, monkeypatch):
|
||
"""Stub-Test: Votes werden korrekt geparst."""
|
||
import httpx
|
||
from app import abgeordnetenwatch as aw_mod
|
||
|
||
fake_votes = [
|
||
{
|
||
"mandate": {
|
||
"id": 101,
|
||
"label": "Erika Mustermann",
|
||
"party": {"label": "SPD"},
|
||
},
|
||
"vote": "yes",
|
||
},
|
||
{
|
||
"mandate": {
|
||
"id": 102,
|
||
"label": "Max Muster",
|
||
"party": {"label": "CDU"},
|
||
},
|
||
"vote": "no",
|
||
},
|
||
]
|
||
|
||
class FakeResponse:
|
||
def raise_for_status(self): pass
|
||
def json(self): return {"data": fake_votes}
|
||
|
||
class FakeClient:
|
||
async def __aenter__(self): return self
|
||
async def __aexit__(self, *a): pass
|
||
async def get(self, url, params=None): return FakeResponse()
|
||
|
||
monkeypatch.setattr(httpx, "AsyncClient", lambda **kw: FakeClient())
|
||
|
||
votes = run(aw_mod.fetch_votes_for_poll(42))
|
||
assert len(votes) == 2
|
||
assert votes[0]["politician_id"] == 101
|
||
assert votes[0]["vote"] == "yes"
|
||
assert votes[0]["partei"] == "SPD"
|
||
assert votes[1]["vote"] == "no"
|
||
|
||
def test_fetch_votes_unknown_vote_value_becomes_no_show(self, monkeypatch):
|
||
import httpx
|
||
from app import abgeordnetenwatch as aw_mod
|
||
|
||
fake_votes = [{"mandate": {"id": 1, "label": "X"}, "vote": "gibberish"}]
|
||
|
||
class FakeResponse:
|
||
def raise_for_status(self): pass
|
||
def json(self): return {"data": fake_votes}
|
||
|
||
class FakeClient:
|
||
async def __aenter__(self): return self
|
||
async def __aexit__(self, *a): pass
|
||
async def get(self, url, params=None): return FakeResponse()
|
||
|
||
monkeypatch.setattr(httpx, "AsyncClient", lambda **kw: FakeClient())
|
||
|
||
votes = run(aw_mod.fetch_votes_for_poll(99))
|
||
assert votes[0]["vote"] == "no_show"
|
||
|
||
|
||
# ─── DB-Upsert-Round-Trip ─────────────────────────────────────────────────────
|
||
|
||
class TestDbUpsertRoundTrip:
|
||
def test_upsert_poll_new(self, initialized_db):
|
||
from app import database
|
||
is_new = run(database.upsert_aw_poll(
|
||
poll_id=1001,
|
||
parliament_id=4,
|
||
bundesland="NRW",
|
||
drucksache="18/500",
|
||
titel="Test-Abstimmung",
|
||
datum="2026-04-01",
|
||
accepted=True,
|
||
topics=["Klimaschutz"],
|
||
legislature_label="18. Wahlperiode",
|
||
synced_at="2026-04-20T10:00:00",
|
||
))
|
||
assert is_new is True
|
||
|
||
def test_upsert_poll_update_returns_false(self, initialized_db):
|
||
from app import database
|
||
run(database.upsert_aw_poll(
|
||
poll_id=1002, parliament_id=4, bundesland="NRW",
|
||
drucksache="18/501", titel="Alt", datum="2026-03-01",
|
||
accepted=False, topics=[], legislature_label="",
|
||
synced_at="2026-04-20T10:00:00",
|
||
))
|
||
is_new = run(database.upsert_aw_poll(
|
||
poll_id=1002, parliament_id=4, bundesland="NRW",
|
||
drucksache="18/501", titel="Neu", datum="2026-03-01",
|
||
accepted=True, topics=[], legislature_label="",
|
||
synced_at="2026-04-20T11:00:00",
|
||
))
|
||
assert is_new is False
|
||
|
||
def test_upsert_vote_new(self, initialized_db):
|
||
from app import database
|
||
run(database.upsert_aw_poll(
|
||
poll_id=2001, parliament_id=4, bundesland="NRW",
|
||
drucksache=None, titel="V-Test", datum="2026-04-01",
|
||
accepted=True, topics=[], legislature_label="",
|
||
synced_at="2026-04-20T10:00:00",
|
||
))
|
||
is_new = run(database.upsert_aw_vote(
|
||
poll_id=2001, politician_id=999,
|
||
politician_name="Test Politiker",
|
||
partei="SPD",
|
||
vote="yes",
|
||
))
|
||
assert is_new is True
|
||
|
||
def test_upsert_vote_update_returns_false(self, initialized_db):
|
||
from app import database
|
||
run(database.upsert_aw_poll(
|
||
poll_id=2002, parliament_id=4, bundesland="NRW",
|
||
drucksache=None, titel="V-Test2", datum="2026-04-01",
|
||
accepted=True, topics=[], legislature_label="",
|
||
synced_at="2026-04-20T10:00:00",
|
||
))
|
||
run(database.upsert_aw_vote(2002, 888, "Name", "CDU", "no"))
|
||
is_new = run(database.upsert_aw_vote(2002, 888, "Name", "CDU", "yes"))
|
||
assert is_new is False
|
||
|
||
def test_get_abstimmungsverhalten_returns_none_for_missing(self, initialized_db):
|
||
from app import database
|
||
result = run(database.get_abstimmungsverhalten("99/9999"))
|
||
assert result is None
|
||
|
||
def test_get_abstimmungsverhalten_aggregates(self, initialized_db):
|
||
from app import database
|
||
# Poll anlegen
|
||
run(database.upsert_aw_poll(
|
||
poll_id=3001, parliament_id=4, bundesland="NRW",
|
||
drucksache="18/3001", titel="AggTest", datum="2026-04-10",
|
||
accepted=True, topics=[], legislature_label="",
|
||
synced_at="2026-04-20T10:00:00",
|
||
))
|
||
# Votes: 2× SPD yes, 1× CDU no, 1× CDU abstain
|
||
run(database.upsert_aw_vote(3001, 1, "A", "SPD", "yes"))
|
||
run(database.upsert_aw_vote(3001, 2, "B", "SPD", "yes"))
|
||
run(database.upsert_aw_vote(3001, 3, "C", "CDU", "no"))
|
||
run(database.upsert_aw_vote(3001, 4, "D", "CDU", "abstain"))
|
||
|
||
result = run(database.get_abstimmungsverhalten("18/3001"))
|
||
assert result is not None
|
||
assert result["accepted"] is True
|
||
fraktionen = {f["partei"]: f for f in result["fraktionen"]}
|
||
assert fraktionen["SPD"]["yes"] == 2
|
||
assert fraktionen["CDU"]["no"] == 1
|
||
assert fraktionen["CDU"]["abstain"] == 1
|
||
|
||
def test_aw_tables_created_by_init_db(self, db_path):
|
||
"""init_db legt abgeordnetenwatch_polls und _votes an."""
|
||
import aiosqlite
|
||
from app import database
|
||
run(database.init_db())
|
||
|
||
async def check():
|
||
async with aiosqlite.connect(db_path) as db:
|
||
cur = await db.execute(
|
||
"SELECT name FROM sqlite_master WHERE type='table'"
|
||
)
|
||
return {r[0] for r in await cur.fetchall()}
|
||
|
||
tables = run(check())
|
||
assert "abgeordnetenwatch_polls" in tables
|
||
assert "abgeordnetenwatch_votes" in tables
|
||
|
||
|
||
# ─── fallback_drucksache_by_date_title (#142 Phase 3) ────────────────────────
|
||
|
||
class TestFallbackDrucksacheByDateTitle:
|
||
"""Unit-Tests für den Datum+Titel-Fallback-Lookup gegen die Assessments-DB."""
|
||
|
||
@pytest.fixture()
|
||
def db_with_assessment(self, initialized_db):
|
||
"""DB mit einem vorbereiteten Assessment für den Fallback-Test."""
|
||
from app import database
|
||
run(database.upsert_assessment({
|
||
"drucksache": "7/1234",
|
||
"title": "Antrag zur Verbesserung des Nahverkehrs in MV",
|
||
"bundesland": "MV",
|
||
"datum": "2026-04-10",
|
||
"fraktionen": ["SPD"],
|
||
"gwoeScore": 7.0,
|
||
"gwoeBegründung": "Test",
|
||
"gwoeMatrix": [],
|
||
"gwoeSchwerpunkt": [],
|
||
"wahlprogrammScores": [],
|
||
"verbesserungen": [],
|
||
"stärken": [],
|
||
"schwächen": [],
|
||
"themen": [],
|
||
"antragZusammenfassung": "",
|
||
"antragKernpunkte": [],
|
||
"source": "batch",
|
||
"model": "test",
|
||
}))
|
||
return initialized_db
|
||
|
||
def test_fallback_finds_match_within_14_days(self, db_with_assessment):
|
||
from app.abgeordnetenwatch import fallback_drucksache_by_date_title
|
||
result = run(fallback_drucksache_by_date_title(
|
||
datum="2026-04-12", # 2 Tage nach Assessment-Datum
|
||
titel="Verbesserung des Nahverkehrs in MV",
|
||
bundesland="MV",
|
||
))
|
||
assert result == "7/1234"
|
||
|
||
def test_fallback_returns_none_outside_14_days(self, db_with_assessment):
|
||
from app.abgeordnetenwatch import fallback_drucksache_by_date_title
|
||
result = run(fallback_drucksache_by_date_title(
|
||
datum="2026-05-01", # 21 Tage nach Assessment-Datum
|
||
titel="Verbesserung des Nahverkehrs in MV",
|
||
bundesland="MV",
|
||
))
|
||
assert result is None
|
||
|
||
def test_fallback_returns_none_wrong_bundesland(self, db_with_assessment):
|
||
from app.abgeordnetenwatch import fallback_drucksache_by_date_title
|
||
result = run(fallback_drucksache_by_date_title(
|
||
datum="2026-04-10",
|
||
titel="Verbesserung des Nahverkehrs in MV",
|
||
bundesland="BY", # falsches BL
|
||
))
|
||
assert result is None
|
||
|
||
def test_fallback_returns_none_no_titel_match(self, db_with_assessment):
|
||
from app.abgeordnetenwatch import fallback_drucksache_by_date_title
|
||
result = run(fallback_drucksache_by_date_title(
|
||
datum="2026-04-10",
|
||
titel="Irgendwas voellig anderes ohne Treffer",
|
||
bundesland="MV",
|
||
))
|
||
assert result is None
|
||
|
||
def test_fallback_returns_none_for_missing_inputs(self, db_with_assessment):
|
||
from app.abgeordnetenwatch import fallback_drucksache_by_date_title
|
||
assert run(fallback_drucksache_by_date_title(None, "Titel", "MV")) is None
|
||
assert run(fallback_drucksache_by_date_title("2026-04-10", None, "MV")) is None
|