"""Tests for the Research Planner — Stage 4.""" from __future__ import annotations from typing import Any import pytest from nsct.agents.planner import MockResearchPlanner, ResearchPlanner from nsct.agents.validator import validate_plan from nsct.config import AppSettings # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- def _make_mock_planner() -> MockResearchPlanner: """Erzeuge einen Mock-Planner für Tests.""" return MockResearchPlanner() def _valid_plan() -> dict[str, Any]: """Erzeuge einen gültigen Plan-Dict (ohne LLM).""" return { "topic": "Test-Thema", "time_range": { "start": None, "end": None, "description": "Ganzer Zeitraum", }, "entities": ["Bundesregierung", "Opposition", "EU"], "search_dimensions": [ "primary_sources", "independent_reporting", "counter_evidence", "scientific_sources", ], "queries": [ { "query": "Was ist Test-Thema?", "purpose": "Allgemeines Verständnis", "category": "general", "language": "de", }, { "query": "Test-Thema offizielle Daten", "purpose": "Primärquellen", "category": "primary_source", "language": "de", }, { "query": "Test-Thema aktuelle Nachrichten", "purpose": "Aktuelle Berichterstattung", "category": "news", "language": "de", }, { "query": "Test-Thema Kritik Kontroverse", "purpose": "Gegenstimmen", "category": "counter_evidence", "language": "de", }, { "query": "Test-Thema wissenschaftliche Analyse", "purpose": "Wissenschaftliche Quellen", "category": "scientific", "language": "de", }, { "query": "Test-Thema unabhängige Bewertung", "purpose": "Unabhängige Quellen", "category": "general", "language": "de", }, ], "potential_sources": [ {"type": "primary_source", "description": "Behörden-Websites"}, {"type": "secondary_source", "description": "Nachrichtenagenturen"}, {"type": "academic", "description": "Wissenschaftliche Datenbanken"}, ], "counter_hypotheses": [ "Alternative Interpretation 1", "Alternative Interpretation 2", ], "search_bias_mitigation": [ "Neutrale Suchbegriffe nutzen", "Mehrere Quellen vergleichen", ], "estimated_depth": "normal", "confidence": 0.8, } def _invalid_plan() -> dict[str, Any]: """Erzeuge einen ungültigen Plan-Dict (für validate_plan-Tests).""" return { "topic": "", "time_range": {}, "entities": [], "search_dimensions": [], "queries": [], "potential_sources": [], "counter_hypotheses": [], "search_bias_mitigation": [], "estimated_depth": "invalid", "confidence": 2.0, } # --------------------------------------------------------------------------- # Tests: Valid Plan hat alle required fields # --------------------------------------------------------------------------- def test_valid_plan_has_required_fields() -> None: """Ein gültiger Plan muss alle required fields enthalten.""" plan = _valid_plan() required = [ "topic", "time_range", "entities", "search_dimensions", "queries", "potential_sources", "counter_hypotheses", "search_bias_mitigation", "estimated_depth", "confidence", ] for field in required: assert field in plan, f"Fehlendes required field: {field}" # --------------------------------------------------------------------------- # Tests: Mindestens 4 verschiedene query categories # --------------------------------------------------------------------------- def test_min_four_query_categories() -> None: """Es müssen mindestens 4 verschiedene query categories vorhanden sein.""" plan = _valid_plan() categories = {q["category"] for q in plan["queries"]} assert len(categories) >= 4, ( f"Nur {len(categories)} categories gefunden: {sorted(categories)}" ) # Die gültige Test-Plan sollte 4 unique haben: general, primary_source, # news, counter_evidence, scientific (5 unique) assert len(categories) >= 4 # --------------------------------------------------------------------------- # Tests: counter_evidence query vorhanden # --------------------------------------------------------------------------- def test_counter_evidence_query_present() -> None: """Es muss mindestens eine query mit category 'counter_evidence' geben.""" plan = _valid_plan() categories = [q["category"] for q in plan["queries"]] assert "counter_evidence" in categories, ( "Keine query mit category 'counter_evidence' gefunden" ) # --------------------------------------------------------------------------- # Tests: search_dimensions enthält primary_sources und counter_evidence # --------------------------------------------------------------------------- def test_search_dimensions_required() -> None: """search_dimensions muss 'primary_sources' und 'counter_evidence' enthalten.""" plan = _valid_plan() dims = plan["search_dimensions"] assert "primary_sources" in dims, "search_dimensions fehlt 'primary_sources'" assert "counter_evidence" in dims, "search_dimensions fehlt 'counter_evidence'" # --------------------------------------------------------------------------- # Tests: validate_plan mit gültigem Plan # --------------------------------------------------------------------------- def test_validate_plan_valid() -> None: """validate_plan soll einen gültigen Plan als gültig erkennen.""" plan = _valid_plan() result = validate_plan(plan) assert result["valid"] is True assert result["errors"] == [] def test_validate_plan_invalid() -> None: """validate_plan soll einen ungültigen Plan als ungültig erkennen.""" plan = _invalid_plan() result = validate_plan(plan) assert result["valid"] is False assert len(result["errors"]) > 0 # --------------------------------------------------------------------------- # Tests: confidence im Bereich 0.0–1.0 # --------------------------------------------------------------------------- def test_valid_plan_confidence_in_range() -> None: """confidence eines gültigen Plans muss im Bereich 0.0–1.0 sein.""" plan = _valid_plan() assert 0.0 <= plan["confidence"] <= 1.0, ( f"confidence {plan['confidence']} außerhalb des Bereichs [0.0, 1.0]" ) def test_validate_plan_confidence_out_of_range() -> None: """validate_plan soll confidence > 1.0 ablehnen.""" plan = _valid_plan() plan["confidence"] = 1.5 result = validate_plan(plan) assert result["valid"] is False assert any("confidence" in err for err in result["errors"]) def test_validate_plan_confidence_negative() -> None: """validate_plan soll confidence < 0.0 ablehnen.""" plan = _valid_plan() plan["confidence"] = -0.1 result = validate_plan(plan) assert result["valid"] is False assert any("confidence" in err for err in result["errors"]) # --------------------------------------------------------------------------- # Tests: Search-Bias-Mitigation (politische Anfrage → beide Seiten) # --------------------------------------------------------------------------- def test_search_bias_both_sides_covered() -> None: """Politische Anfragen müssen beide Seiten abdecken.""" planner = _make_mock_planner() plan = planner.plan( research_question="Umweltpolitik der Bundesregierung 2024", language="de", ) # search_dimensions muss counter_evidence enthalten assert "counter_evidence" in plan["search_dimensions"] # Es muss eine counter_evidence-Query geben has_counter = any( q["category"] == "counter_evidence" for q in plan["queries"] ) assert has_counter, "Keine counter_evidence-Query in der generierten Strategie" # Es muss general/neutral Queries geben (andere Seite) has_general = any( q["category"] == "general" for q in plan["queries"] ) assert has_general, "Keine general-Query in der generierten Strategie" # Mindestens 6 queries für vollständige Bias-Mitigation assert len(plan["queries"]) >= 6, ( f"Nur {len(plan['queries'])} queries — mindestens 6 für vollständige Bias-Mitigation" ) # search_bias_mitigation muss nicht leer sein assert len(plan["search_bias_mitigation"]) >= 2, ( "search_bias_mitigation sollte mindestens 2 Maßnahmen enthalten" ) # counter_hypotheses muss nicht leer sein assert len(plan["counter_hypotheses"]) >= 1, ( "counter_hypotheses sollte mindestens 1 Eintrag enthalten" ) def test_search_bias_neutral_queries() -> None: """Suchanfragen müssen neutrale Formulierungen verwenden, nicht einseitig.» Verifiziert, dass die generierten queries nicht nur einseitig politische Begriffe enthalten. """ planner = _make_mock_planner() plan = planner.plan( research_question="Flüchtling政策 und Integration in Deutschland", language="de", ) # Alle Queries müssen purpose und category haben for q in plan["queries"]: assert q.get("query", "").strip(), "Query darf nicht leer sein" assert q.get("purpose", "").strip(), "purpose darf nicht leer sein" assert q.get("category"), "category darf nicht leer sein" # Es muss sowohl supporting als auch counter_evidence geben categories = [q["category"] for q in plan["queries"]] assert "counter_evidence" in categories assert "news" in categories or "general" in categories # --------------------------------------------------------------------------- # Tests: MockResearchPlanner generiert validen Plan # --------------------------------------------------------------------------- def test_mock_planner_generates_valid_plan() -> None: """Der Mock-Planner muss einen validen Plan generieren.""" planner = _make_mock_planner() plan = planner.plan( research_question="Test-Frage zu Klimapolitik", language="de", ) # Alle required fields vorhanden? result = validate_plan(plan) assert result["valid"] is True, f"Validation errors: {result['errors']}" # topic nicht leer assert plan["topic"].strip() # search_dimensions muss korrekt sein assert len(plan["search_dimensions"]) > 0 # Mindestens 6 queries assert len(plan["queries"]) >= 6 def test_mock_planner_detects_depth_quick() -> None: """Depth-Erkennung für kurze Anfragen.""" planner = _make_mock_planner() plan = planner.plan( research_question="Was ist 2+2? Kurzantwort.", language="de", ) assert plan["estimated_depth"] == "quick" def test_mock_planner_detects_depth_normal() -> None: """Depth-Erkennung für normale Anfragen.""" planner = _make_mock_planner() plan = planner.plan( research_question="Stand der Elektromobilität in Deutschland", language="de", ) assert plan["estimated_depth"] == "normal" def test_mock_planner_detects_depth_deep() -> None: """Depth-Erkennung für tiefgehende Anfragen.""" planner = _make_mock_planner() plan = planner.plan( research_question="Tiefgehende Analyse der deutschen Energiewende und deren Auswirkungen", language="de", ) assert plan["estimated_depth"] == "deep" # --------------------------------------------------------------------------- # Tests: API-Endpoint (Mock, kein echtes LLM) # --------------------------------------------------------------------------- def test_planner_endpoint_returns_valid_plan() -> None: """POST /research/planner muss einen gültigen Plan zurückgeben.""" from nsct.api.main import create_app from fastapi.testclient import TestClient app = create_app() with TestClient(app) as client: resp = client.post( "/research/planner", json={"query": "Test-Frage", "language": "de"}, ) assert resp.status_code == 200 body = resp.json() assert "plan" in body assert "valid" in body assert isinstance(body["plan"], dict) assert isinstance(body["valid"], bool) def test_planner_endpoint_minimal_query() -> None: """POST /research/planner mit minimaler Anfrage.""" from nsct.api.main import create_app from fastapi.testclient import TestClient app = create_app() with TestClient(app) as client: resp = client.post("/research/planner", json={"query": "Test"}) assert resp.status_code == 200 body = resp.json() assert body["valid"] is True def test_planner_endpoint_422_on_empty_query() -> None: """POST /research/planner mit leerer query muss 422 zurückgeben.""" from nsct.api.main import create_app from fastapi.testclient import TestClient app = create_app() with TestClient(app) as client: resp = client.post("/research/planner", json={"query": ""}) assert resp.status_code == 422 # --------------------------------------------------------------------------- # Tests: ResearchPlan Pydantic-Schema (models/plan.py) # --------------------------------------------------------------------------- def test_research_plan_schema_valid() -> None: """ResearchPlan muss mit allen gültigen Feldern instanziert werden.""" from nsct.models.plan import ( QueryConfig, PotentialSource, ResearchPlan, TimeRange, ) plan = ResearchPlan( topic="Test-Thema", time_range=TimeRange(start=None, end=None, description="Test"), entities=["Entität1"], search_dimensions=["primary_sources", "counter_evidence"], queries=[ QueryConfig( query="Test query", purpose="Test", category="general", language="de", ), ], potential_sources=[ PotentialSource(type="primary_source", description="Test"), ], counter_hypotheses=["Hypothese 1"], search_bias_mitigation=["Mitigation 1"], estimated_depth="normal", confidence=0.7, ) assert plan.topic == "Test-Thema" assert plan.confidence == 0.7 def test_research_plan_schema_confidence_bounds() -> None: """ResearchPlan muss confidence 0.0 und 1.0 erlauben.""" from nsct.models.plan import ( QueryConfig, PotentialSource, ResearchPlan, TimeRange, ) # confidence = 0.0 plan_min = ResearchPlan( topic="T", time_range=TimeRange(start=None, end=None, description="T"), entities=["E"], search_dimensions=["primary_sources"], queries=[QueryConfig(query="q", purpose="p", category="general", language="de")], potential_sources=[PotentialSource(type="primary_source", description="d")], confidence=0.0, ) assert plan_min.confidence == 0.0 # confidence = 1.0 plan_max = ResearchPlan( topic="T", time_range=TimeRange(start=None, end=None, description="T"), entities=["E"], search_dimensions=["primary_sources"], queries=[QueryConfig(query="q", purpose="p", category="general", language="de")], potential_sources=[PotentialSource(type="primary_source", description="d")], confidence=1.0, ) assert plan_max.confidence == 1.0 # --------------------------------------------------------------------------- # Tests: validator edge cases # --------------------------------------------------------------------------- def test_validate_missing_fields() -> None: """validate_plan soll fehlende Felder melden.""" result = validate_plan({}) assert result["valid"] is False assert len(result["errors"]) > 0 def test_validate_missing_counter_evidence() -> None: """validate_plan soll fehlende counter_evidence-Query melden.""" plan = _valid_plan() # Entferne die counter_evidence-Query plan["queries"] = [ q for q in plan["queries"] if q["category"] != "counter_evidence" ] result = validate_plan(plan) assert result["valid"] is False assert any("counter_evidence" in err for err in result["errors"]) def test_validate_missing_search_dimensions() -> None: """validate_plan soll fehlende search_dimensions melden.""" plan = _valid_plan() plan["search_dimensions"] = [] result = validate_plan(plan) assert result["valid"] is False assert any("primary_sources" in err for err in result["errors"]) def test_validate_empty_topic() -> None: """validate_plan soll leeres topic melden.""" plan = _valid_plan() plan["topic"] = "" result = validate_plan(plan) assert result["valid"] is False # --------------------------------------------------------------------------- # Tests: Mock Planner search_dimensions completeness # --------------------------------------------------------------------------- def test_mock_planner_search_dimensions() -> None: """Der Mock-Planner muss alle 4 search_dimensions setzen.""" planner = _make_mock_planner() plan = planner.plan(research_question="Test", language="de") required_dims = {"primary_sources", "independent_reporting", "counter_evidence", "scientific_sources"} found = set(plan["search_dimensions"]) assert required_dims.issubset(found), ( f"search_dimensions fehlen: {required_dims - found}" )