"""Tests fuer app.presse_generator (#170 Phase 4).""" from __future__ import annotations import json import sqlite3 from pathlib import Path from unittest.mock import patch import pytest from app.presse_generator import ( _build_user_prompt, _find_existing_draft, generate_draft, get_draft, list_drafts, ) # ───────────────────────────────────────────────────────────────────────────── # Fixture: DB mit Antrag + News # ───────────────────────────────────────────────────────────────────────────── @pytest.fixture def db_with_antrag_and_news(tmp_path: Path) -> Path: db = tmp_path / "test_presse.db" conn = sqlite3.connect(str(db)) conn.execute(""" CREATE TABLE assessments ( drucksache TEXT PRIMARY KEY, title TEXT, bundesland TEXT, antrag_zusammenfassung TEXT, gwoe_score REAL, gwoe_begruendung TEXT, empfehlung TEXT ) """) conn.execute(""" CREATE TABLE news_articles ( url TEXT PRIMARY KEY, titel TEXT NOT NULL, summary TEXT ) """) conn.execute(""" CREATE TABLE presse_drafts ( id INTEGER PRIMARY KEY AUTOINCREMENT, drucksache TEXT NOT NULL, bundesland TEXT NOT NULL, news_url TEXT NOT NULL, news_titel TEXT NOT NULL, titel TEXT NOT NULL, body TEXT NOT NULL, model TEXT NOT NULL, style TEXT NOT NULL DEFAULT 'pm', created_at TEXT NOT NULL DEFAULT (datetime('now')) ) """) conn.execute( """INSERT INTO assessments (drucksache, title, bundesland, antrag_zusammenfassung, gwoe_score, gwoe_begruendung, empfehlung) VALUES (?, ?, ?, ?, ?, ?, ?)""", ( "18/A", "Wohnungsbau-Reform-Antrag", "NRW", "Antrag fuer mehr sozialen Wohnungsbau", 8.5, "Stark gemeinwohlorientiert", "Uneingeschränkt unterstützen", ), ) conn.execute( "INSERT INTO news_articles (url, titel, summary) VALUES (?, ?, ?)", ( "https://example.com/wohnen", "Wohnungsmarkt im Umbruch", "Die Mietpreise steigen weiter, der Bundestag berät heute", ), ) conn.commit() conn.close() return db # ───────────────────────────────────────────────────────────────────────────── # _build_user_prompt # ───────────────────────────────────────────────────────────────────────────── class TestBuildUserPrompt: def test_includes_drucksache(self): prompt = _build_user_prompt( drucksache="18/A", bundesland="NRW", antrag_titel="Test", antrag_zusammenfassung="Summary", gwoe_score=7.5, gwoe_begruendung="ok", empfehlung="Unterstützen", news_titel="News", news_summary="Lead", news_url="https://example.com", ) assert "18/A" in prompt assert "NRW" in prompt assert "7.5" in prompt assert "News" in prompt def test_handles_missing_zusammenfassung(self): prompt = _build_user_prompt( drucksache="x", bundesland="x", antrag_titel="x", antrag_zusammenfassung="", gwoe_score=5.0, gwoe_begruendung="", empfehlung="", news_titel="x", news_summary="", news_url="", ) assert "(keine vorhanden)" in prompt # ───────────────────────────────────────────────────────────────────────────── # generate_draft (mocked QwenBewerter) # ───────────────────────────────────────────────────────────────────────────── class FakeBewerter: """Mock fuer QwenBewerter, gibt fixe LLM-Response zurueck.""" def __init__(self, response: dict): self._response = response self.last_request = None async def bewerte(self, request): self.last_request = request return self._response @pytest.mark.asyncio async def test_generate_draft_persists_record(db_with_antrag_and_news, monkeypatch): bewerter = FakeBewerter({ "titel": "Wohnungsbau jetzt", "body": "Der vorliegende Antrag der Drucksache 18/A ..." * 10, # langer Body }) # Patch settings.dashscope_model fuer den INSERT from app.config import settings as real_settings monkeypatch.setattr(real_settings, "llm_model_premium", "qwen-test") result = await generate_draft( drucksache="18/A", news_url="https://example.com/wohnen", db_path=db_with_antrag_and_news, bewerter=bewerter, ) assert result["id"] == 1 assert result["drucksache"] == "18/A" assert result["bundesland"] == "NRW" assert result["news_titel"] == "Wohnungsmarkt im Umbruch" assert result["titel"] == "Wohnungsbau jetzt" assert "18/A" in result["body"] assert result["_was_existing"] is False assert result["model"] == "qwen-test" # premium-Modell wurde verwendet @pytest.mark.asyncio async def test_generate_draft_idempotency_returns_existing( db_with_antrag_and_news, monkeypatch, ): """Zweiter Aufruf mit gleicher (drucksache, news_url) liefert den existing Draft, ohne neuen LLM-Call (kein call_count increase).""" bewerter = FakeBewerter({ "titel": "PM Erstgenerierung", "body": "Body 1 ..." * 30, }) from app.config import settings as real_settings monkeypatch.setattr(real_settings, "llm_model_premium", "qwen-test") first = await generate_draft( drucksache="18/A", news_url="https://example.com/wohnen", db_path=db_with_antrag_and_news, bewerter=bewerter, ) assert first["_was_existing"] is False # Zweiter Call mit anderer Bewerter-Antwort, soll aber NICHT verwendet # werden — existing Draft wird zurueckgeliefert. bewerter2 = FakeBewerter({ "titel": "Sollte NICHT auftauchen", "body": "Sollte NICHT auftauchen ..." * 30, }) second = await generate_draft( drucksache="18/A", news_url="https://example.com/wohnen", db_path=db_with_antrag_and_news, bewerter=bewerter2, ) assert second["_was_existing"] is True assert second["id"] == first["id"] assert second["titel"] == "PM Erstgenerierung" # nicht der zweite! # Bewerter2 wurde NICHT aufgerufen assert bewerter2.last_request is None @pytest.mark.asyncio async def test_generate_draft_force_makes_new_call( db_with_antrag_and_news, monkeypatch, ): """Mit force=True wird auch bei vorhandenem Draft neu generiert.""" bewerter = FakeBewerter({"titel": "Erstgen", "body": "Body 1 " * 30}) from app.config import settings as real_settings monkeypatch.setattr(real_settings, "llm_model_premium", "qwen-test") first = await generate_draft( drucksache="18/A", news_url="https://example.com/wohnen", db_path=db_with_antrag_and_news, bewerter=bewerter, ) bewerter2 = FakeBewerter({"titel": "Force-Regen", "body": "Body 2 " * 30}) second = await generate_draft( drucksache="18/A", news_url="https://example.com/wohnen", db_path=db_with_antrag_and_news, bewerter=bewerter2, force=True, ) assert second["_was_existing"] is False assert second["id"] != first["id"] assert second["titel"] == "Force-Regen" assert bewerter2.last_request is not None @pytest.mark.asyncio async def test_generate_draft_unknown_drucksache(db_with_antrag_and_news): bewerter = FakeBewerter({"titel": "x", "body": "y"}) with pytest.raises(ValueError, match="Drucksache"): await generate_draft( drucksache="99/MISSING", news_url="https://example.com/wohnen", db_path=db_with_antrag_and_news, bewerter=bewerter, ) @pytest.mark.asyncio async def test_generate_draft_unknown_news(db_with_antrag_and_news): bewerter = FakeBewerter({"titel": "x", "body": "y"}) with pytest.raises(ValueError, match="News-URL"): await generate_draft( drucksache="18/A", news_url="https://example.com/missing", db_path=db_with_antrag_and_news, bewerter=bewerter, ) @pytest.mark.asyncio async def test_generate_draft_empty_response_raises(db_with_antrag_and_news, monkeypatch): bewerter = FakeBewerter({"titel": "", "body": ""}) from app.config import settings as real_settings monkeypatch.setattr(real_settings, "llm_model_premium", "qwen-test") with pytest.raises(ValueError, match="unvollständig"): await generate_draft( drucksache="18/A", news_url="https://example.com/wohnen", db_path=db_with_antrag_and_news, bewerter=bewerter, ) # ───────────────────────────────────────────────────────────────────────────── # list_drafts + get_draft # ───────────────────────────────────────────────────────────────────────────── class TestFindExistingDraft: def test_returns_none_when_no_match(self, db_with_antrag_and_news): assert _find_existing_draft( "18/A", "https://x.de/n", db_with_antrag_and_news, ) is None def test_returns_existing_draft(self, db_with_antrag_and_news): conn = sqlite3.connect(str(db_with_antrag_and_news)) conn.execute( """INSERT INTO presse_drafts (drucksache, bundesland, news_url, news_titel, titel, body, model) VALUES (?, ?, ?, ?, ?, ?, ?)""", ("18/A", "NRW", "https://x.de/n", "News", "PM-Titel", "PM-Body", "qwen-test"), ) conn.commit() conn.close() d = _find_existing_draft("18/A", "https://x.de/n", db_with_antrag_and_news) assert d is not None assert d["titel"] == "PM-Titel" def test_returns_newest_when_multiple_exist(self, db_with_antrag_and_news): conn = sqlite3.connect(str(db_with_antrag_and_news)) for titel in ["Alt1", "Alt2", "Neu"]: conn.execute( """INSERT INTO presse_drafts (drucksache, bundesland, news_url, news_titel, titel, body, model) VALUES (?, ?, ?, ?, ?, ?, ?)""", ("18/A", "NRW", "https://x.de/n", "News", titel, "Body", "qwen-test"), ) conn.commit() conn.close() d = _find_existing_draft("18/A", "https://x.de/n", db_with_antrag_and_news) # Neuester Draft (höchste id) zurueckgeliefert assert d["titel"] == "Neu" class TestListAndGetDrafts: def test_empty(self, db_with_antrag_and_news): assert list_drafts(db_path=db_with_antrag_and_news) == [] assert get_draft(99, db_path=db_with_antrag_and_news) is None def test_after_insert(self, db_with_antrag_and_news): # Direct DB-Insert (test setup) conn = sqlite3.connect(str(db_with_antrag_and_news)) conn.execute( """INSERT INTO presse_drafts (drucksache, bundesland, news_url, news_titel, titel, body, model) VALUES (?, ?, ?, ?, ?, ?, ?)""", ("18/A", "NRW", "https://x.de/n", "News-Titel", "PM-Titel", "PM-Body", "test-model"), ) conn.commit() conn.close() drafts = list_drafts(db_path=db_with_antrag_and_news) assert len(drafts) == 1 assert drafts[0]["drucksache"] == "18/A" assert drafts[0]["titel"] == "PM-Titel" d = get_draft(drafts[0]["id"], db_path=db_with_antrag_and_news) assert d is not None assert d["body"] == "PM-Body"