"""Tests: "find literature" really queries databases, instead of stopping after formulating the search queries. """ import asyncio from dataclasses import dataclass, field import tests.conftest # noqa: F401 (dependency stubs) from src.pipeline.analysis_pipeline import ( AnalysisLayerNode, AnalysisPipelineRunner, Decomposer, _year_from, ) from src.pipeline.models import TaskResult, TaskState from src.pipeline.search_executor import ( MAX_QUERIES, extract_queries, format_results_markdown, results_to_dicts, run_literature_search, ) PREFLIGHT = {"topic": "Hybrides Arbeiten", "discipline": "BWL", "time_window": "2020-2025"} @dataclass class FakeHit: title: str = "Ein Titel" authors: list = field(default_factory=lambda: ["Meier", "Schulz"]) year: int = 2022 venue: str = "Journal of Work" doi: str = "10.1000/abc" url: str = "https://example.org/abc" abstract: str = "Abstract." cited_by_count: int = 12 sources: list = field(default_factory=lambda: ["openalex"]) class FakeService: def __init__(self, per_query=2, fail_on=()): self.per_query = per_query self.fail_on = set(fail_on) self.calls = [] async def multi_source_query(self, query, **kw): self.calls.append((query, kw)) if query in self.fail_on: raise RuntimeError("API down") return [FakeHit(title=f"{query} — Paper {i}", doi=f"10.1/{query}{i}", cited_by_count=10 - i) for i in range(self.per_query)] # ── Query extraction ────────────────────────────────────────────── def test_prefers_structured_queries(): q = extract_queries("egal", {"queries": ["hybrid work", "remote teams"]}) assert q == ["hybrid work", "remote teams"] def test_falls_back_to_lines(): raw = "1. hybrid work\n- remote teams\n* **telework**\n" assert extract_queries(raw, {}) == [ "hybrid work", "remote teams", "telework"] def test_skips_headings_and_prose(): raw = "Suchanfragen:\nhybrid work\n" + "x" * 250 assert extract_queries(raw, {}) == ["hybrid work"] def test_query_count_is_capped(): many = {"queries": [f"q{i}" for i in range(50)]} assert len(extract_queries("", many)) == MAX_QUERIES def test_empty_input_yields_nothing(): assert extract_queries("", {}) == [] assert extract_queries(None, None) == [] # ── Execution ──────────────────────────────────────────────────── def test_search_runs_every_query(): svc = FakeService() md, data = asyncio.run(run_literature_search( ["hybrid work", "remote teams"], svc, {"limit": 5})) assert [c[0] for c in svc.calls] == ["hybrid work", "remote teams"] assert svc.calls[0][1]["limit"] == 5 assert data["n_results"] == 4 assert "hits" in md def test_filters_are_passed_through(): svc = FakeService() asyncio.run(run_literature_search( ["q"], svc, {"limit": 10, "year_from": 2020, "peer_reviewed_only": True})) kw = svc.calls[0][1] assert kw["year_from"] == 2020 and kw["peer_reviewed_only"] is True def test_partial_api_failure_keeps_the_rest(): """One failure must not break the whole search.""" svc = FakeService(fail_on=["kaputt"]) md, data = asyncio.run(run_literature_search( ["gut", "kaputt"], svc, {})) assert data["n_results"] == 2 assert data["errors"] and "1 of 2 queries" in md def test_total_failure_is_reported_clearly(): svc = FakeService(fail_on=["a", "b"]) md, data = asyncio.run(run_literature_search(["a", "b"], svc, {})) assert data["n_results"] == 0 assert "failed" in md def test_no_service_does_not_crash(): md, data = asyncio.run(run_literature_search(["q"], None, {})) assert data["error"] == "no_service" and "not available" in md def test_no_queries_does_not_call_the_api(): svc = FakeService() md, data = asyncio.run(run_literature_search([], svc, {})) assert svc.calls == [] and data["results"] == [] def test_executed_queries_are_shown(): svc = FakeService() md, _ = asyncio.run(run_literature_search(["hybrid work"], svc, {})) assert "`hybrid work`" in md def test_results_are_sorted_by_citations(): svc = FakeService(per_query=3) _, data = asyncio.run(run_literature_search(["q"], svc, {})) cites = [r["cited_by_count"] for r in data["results"]] assert cites == sorted(cites, reverse=True) # ── Formatting ────────────────────────────────────────────────── def test_markdown_contains_verifiable_metadata(): md = format_results_markdown([FakeHit()]) for expected in ("Ein Titel", "2022", "Journal of Work", "10.1000/abc", "12 citations", "Meier"): assert expected in md, expected def test_markdown_handles_empty(): assert "No hits" in format_results_markdown([]) def test_structured_output_keeps_identifiers(): d = results_to_dicts([FakeHit()])[0] assert d["doi"] == "10.1000/abc" and d["year"] == 2022 # ── Time window ─────────────────────────────────────────────────── def test_year_extraction(): assert _year_from("2020-2025") == 2020 assert _year_from("seit 2018") == 2018 assert _year_from("letzte 5 Jahre") is None assert _year_from("") is None # ── The mode as a whole ────────────────────────────────────────── def _plan(): d = Decomposer.__new__(Decomposer) return asyncio.run(d._decompose_literature_finder("Frage", PREFLIGHT)) def test_mode_has_a_real_search_step(): plan = _plan() ids = [t.id for t in plan.tasks] assert ids == ["STRATEGY", "QUERIES", "SEARCH", "ASSESS", "SELECTION"] search = next(t for t in plan.tasks if t.id == "SEARCH") assert search.executor == "literature_search" def test_queries_task_is_structured(): q = next(t for t in _plan().tasks if t.id == "QUERIES") assert q.output_format == "json" def test_time_window_becomes_a_filter(): search = next(t for t in _plan().tasks if t.id == "SEARCH") assert search.search_config["year_from"] == 2020 def test_search_hits_reach_the_report(): """The hit list is the product and must not be dropped.""" plan = _plan() results = { "STRATEGY": TaskResult(task_id="STRATEGY", state=TaskState.DONE, output="Begriffe: hybrid work"), "QUERIES": TaskResult(task_id="QUERIES", state=TaskState.DONE, output='{"queries": ["hybrid work"]}', parsed_output={"queries": ["hybrid work"]}), "SEARCH": TaskResult(task_id="SEARCH", state=TaskState.DONE, output="**2 Treffer**\n\n1. **Ein Titel** (2022)"), "ASSESS": TaskResult(task_id="ASSESS", state=TaskState.DONE, output='{"selected": []}', parsed_output={"selected": []}), "SELECTION": TaskResult(task_id="SELECTION", state=TaskState.DONE, output="Bewertung der Treffer."), } class Ctx: task_plan = plan query = "Frage" preflight_data = PREFLIGHT task_results = results report = AnalysisPipelineRunner._build_final_report(Ctx()) assert "Ein Titel" in report, "hit list missing from the report" assert "Begriffe: hybrid work" in report, "search strategy missing" assert "Bewertung der Treffer." in report assert '{"queries"' not in report, "the JSON intermediate step does not belong in it" assert "## Literature found" in report def test_search_task_reads_queries_from_dependency(): """End to end through the layer node's executor.""" plan = _plan() node = AnalysisLayerNode.__new__(AnalysisLayerNode) node._search_service = FakeService() class Ctx: task_results = { "QUERIES": TaskResult( task_id="QUERIES", state=TaskState.DONE, output='{"queries": ["hybrid work"]}', parsed_output={"queries": ["hybrid work"]}), } search = next(t for t in plan.tasks if t.id == "SEARCH") md, data = asyncio.run(node._run_search_task(search, Ctx())) assert node._search_service.calls[0][0] == "hybrid work" assert data["n_results"] == 2 assert "hits" in md def test_other_modes_keep_single_task_reports(): """No mode except literature_finder may get headings.""" d = Decomposer.__new__(Decomposer) pre = {"decision": "D", "options": "A\nB", "criteria": "K"} plan = asyncio.run(d._decompose_decision_analysis("Q", pre)) assert not any(getattr(t, "show_in_report", False) for t in plan.tasks) # ── Integration with the real data types ───────────────────────── def test_works_with_real_searchresult_objects(): """De-duplication and formatting against the real type, not just stubs.""" from src.connectors.literature_search import SearchResult class RealService: async def multi_source_query(self, query, **kw): return [ SearchResult(id="1", doi="10.1/a", title="Hybrid Work Study", authors=["Meier"], year=2022, venue="JOW", cited_by_count=30, sources=["openalex"]), # duplicate with the same DOI from another source SearchResult(id="2", doi="10.1/a", title="Hybrid Work Study", authors=["Meier"], year=2022, venue="JOW", cited_by_count=30, sources=["s2"]), SearchResult(id="3", doi="10.1/b", title="Remote Teams", authors=["Schulz"], year=2021, cited_by_count=5, sources=["arxiv"]), ] md, data = asyncio.run(run_literature_search( ["hybrid work"], RealService(), {})) assert data["n_results"] == 2, "duplicate was not merged" assert "Hybrid Work Study" in md and "Remote Teams" in md assert data["results"][0]["cited_by_count"] == 30, "not sorted" # ── End to end through the real runner ────────────────────────── class _E2ELLM: """Answers plausibly; `queries` controls the query task.""" def __init__(self, queries=None): self.queries = {"queries": ["hybrid work"]} if queries is None else queries async def primary_complete(self, m, **k): return "### Bewertung\n\n1. **Ein Titel** — einschlägig." async def harvest_complete(self, m, **k): return "Suchbegriffe: hybrid work\nEinschluss: peer-reviewed" async def primary_complete_json(self, m, **k): return self.queries async def harvest_complete_json(self, m, **k): return self.queries class _DeadSearch: async def multi_source_query(self, query, **kw): raise RuntimeError("API unreachable") class _EmptySearch: def __init__(self): self.calls = [] async def multi_source_query(self, query, **kw): self.calls.append(query) return [] def _run(llm, service): from src.pipeline.models import HarvestContext checker = __import__( "src.pipeline.analysis_pipeline", fromlist=["USE_CASE_REGISTRY"] ).USE_CASE_REGISTRY["literature_finder"]["preflight"] runner = AnalysisPipelineRunner( llm_client=llm, progress_callback=lambda *a: asyncio.sleep(0), search_service=service, ) ctx = HarvestContext(query=PREFLIGHT["topic"], use_case="literature_finder") ctx.preflight_data = checker.normalize(PREFLIGHT) async def go(): c = await runner.decompose_only(ctx) return await runner.run(c, skip_decompose=True) return asyncio.run(go()) def test_one_input_yields_a_complete_report(): """The full path: input in, analysed hits out.""" search = FakeService() ctx = _run(_E2ELLM(), search) states = {t.id: ctx.task_results[t.id].state.value for t in ctx.task_plan.tasks} assert states == {"STRATEGY": "done", "QUERIES": "done", "SEARCH": "done", "ASSESS": "done", "SELECTION": "done"}, states assert search.calls, "no database was queried" report = ctx.final_report assert "## Search strategy" in report assert "## Literature found" in report assert "## Assessment of the hits" in report assert "10.1/" in report, "no DOI in the report" assert '{"queries"' not in report def test_runner_accepts_an_injected_search_service(): """Without an entry point, the search could neither be tested nor share an existing API client.""" search = FakeService() runner = AnalysisPipelineRunner( llm_client=_E2ELLM(), progress_callback=None, search_service=search, ) assert runner.search_service is search def test_dead_apis_fail_loudly_and_skip_assessment(): """A failure must not pass as a success. Otherwise the assessment task gets an empty hit list — and invents plausible-sounding titles from it. """ ctx = _run(_E2ELLM(), _DeadSearch()) assert ctx.task_results["SEARCH"].state.value == "failed" assert ctx.task_results["ASSESS"].state.value == "skipped" report = ctx.final_report assert "⚠️" in report and "failed" in report assert "Assessment of the hits" not in report.split("⚠️")[0] def test_zero_hits_is_a_valid_result_not_a_failure(): """A search that ran without hits is not an error.""" ctx = _run(_E2ELLM(), _EmptySearch()) assert ctx.task_results["SEARCH"].state.value == "done" assert ctx.task_results["ASSESS"].state.value == "done" assert "No hits" in ctx.final_report def test_broken_query_task_falls_back_to_the_topic(): """Empty or wrongly structured JSON must not break the run.""" for broken in ({}, {"quatsch": 1}, {"queries": []}): search = _EmptySearch() ctx = _run(_E2ELLM(queries=broken), search) assert search.calls == [PREFLIGHT["topic"]], broken assert ctx.task_results["SEARCH"].state.value == "done" def test_json_text_is_never_line_parsed_into_queries(): """Otherwise `"quatsch": 1` would reach the API as a search query.""" assert extract_queries('{\n "quatsch": 1\n}', {}) == [] assert extract_queries('[1, 2]', {}) == [] assert extract_queries("hybrid work\nremote teams", {}) == [ "hybrid work", "remote teams"] # ── Citation-Chasing ────────────────────────────────────────────── class _ChasingService(FakeService): def __init__(self, **kw): super().__init__(**kw) self.seeds = None self.max_per_seed = None async def multi_source_query(self, query, **kw): from src.connectors.literature_search import SearchResult self.calls.append((query, kw)) return [ SearchResult(id="a", doi="10.1/a", title="Viel zitiert", year=2022, cited_by_count=99, sources=["openalex"]), SearchResult(id="b", doi="10.1/b", title="Wenig zitiert", year=2021, cited_by_count=3, sources=["s2"]), SearchResult(id="c", doi="", title="Ohne DOI", year=2020, cited_by_count=50, sources=["arxiv"]), ] async def expand_from_seeds(self, seed_dois, max_per_seed=20, directions=None): from src.connectors.literature_search import SearchResult self.seeds = list(seed_dois) self.max_per_seed = max_per_seed return [SearchResult(id="z", doi="10.9/z", title="Zitierte Vorarbeit", year=2018, cited_by_count=120, sources=["openalex"])] def test_citation_chasing_only_on_request(): svc = _ChasingService() asyncio.run(run_literature_search(["q"], svc, {})) assert svc.seeds is None, "chasing ran without being asked — it costs 30-90s" def test_citation_chasing_uses_most_cited_dois_as_seeds(): svc = _ChasingService() _, data = asyncio.run(run_literature_search( ["q"], svc, {"expand_citations": True, "max_seeds": 2, "max_per_seed": 7})) assert svc.seeds == ["10.1/a", "10.1/b"], svc.seeds assert svc.max_per_seed == 7 assert data["n_expanded"] == 1 assert any(r["title"] == "Zitierte Vorarbeit" for r in data["results"]) def test_seed_selection_skips_entries_without_doi(): """A hit without a DOI is useless as a seed.""" svc = _ChasingService() asyncio.run(run_literature_search( ["q"], svc, {"expand_citations": True, "max_seeds": 5})) assert "Ohne DOI" not in str(svc.seeds) assert len(svc.seeds) == 2 def test_failed_chasing_keeps_the_query_results(): class Broken(_ChasingService): async def expand_from_seeds(self, *a, **kw): raise RuntimeError("Rate limit") svc = Broken() md, data = asyncio.run(run_literature_search( ["q"], svc, {"expand_citations": True})) assert data["n_results"] == 3, "query hits were lost" assert any("citation_chasing" in e for e in data["errors"]) def test_chasing_is_reported_in_the_markdown(): svc = _ChasingService() md, _ = asyncio.run(run_literature_search( ["q"], svc, {"expand_citations": True})) assert "citation chasing" in md # ── Literature review and grant proposal use the search ─────────────────────────────────── def _plan_for(use_case, preflight): d = Decomposer.__new__(Decomposer) return asyncio.run(getattr(d, f"_decompose_{use_case}")("Q", preflight)) REVIEW_PRE = {"topic": "Hybrides Arbeiten", "discipline": "BWL", "time_window": "2020-2025", "key_questions": "Wie definiert?\nWelche Debatten?"} GRANT_PRE = {"research_question": "Wie wirkt X?", "funder": "DFG", "duration": "3 Jahre", "discipline": "BWL"} def test_literature_review_searches_before_synthesising(): """A review without literature read is just the model's memory.""" plan = _plan_for("literature_review", REVIEW_PRE) ids = [t.id for t in plan.tasks] assert "SEARCH" in ids and "QUERIES" in ids search = next(t for t in plan.tasks if t.id == "SEARCH") assert search.executor == "literature_search" assert search.search_config["expand_citations"] is True # every question synthesis must see the hits for t in plan.tasks: if t.id.startswith("Q_"): assert t.param_deps.get("hits") == "SEARCH", t.id assert "SEARCH" in t.depends_on def test_review_questions_may_now_cite_sources(): """The prompt may ask for titles now that real literature is available.""" plan = _plan_for("literature_review", REVIEW_PRE) q = next(t for t in plan.tasks if t.id.startswith("Q_")) assert "{hits}" in q.prompt_template assert "not in the list" in q.prompt_template def test_grant_proposal_backs_the_state_of_the_art_with_literature(): plan = _plan_for("grant_proposal", GRANT_PRE) ids = [t.id for t in plan.tasks] assert "SEARCH" in ids background = next(t for t in plan.tasks if t.id == "SEC_background") assert background.param_deps.get("hits") == "SEARCH" # the work plan needs no literature workplan = next(t for t in plan.tasks if t.id == "SEC_workplan") assert "hits" not in workplan.param_deps def test_grant_proposal_sections_reach_the_report(): """The report must contain the proposal sections, not only the coherence check.""" plan = _plan_for("grant_proposal", GRANT_PRE) shown = [t.id for t in plan.tasks if t.show_in_report] for key in ("abstract", "background", "methodology", "workplan", "impact"): assert f"SEC_{key}" in shown, key assert "SEARCH" in shown def test_report_order_puts_the_proposal_before_the_bibliography(): plan = _plan_for("grant_proposal", GRANT_PRE) order = [t.id for t in plan.tasks if t.show_in_report] assert order.index("SEC_impact") < order.index("SEARCH") def test_both_modes_stay_acyclic_and_resolvable(): for uc, pre in (("literature_review", REVIEW_PRE), ("grant_proposal", GRANT_PRE)): plan = _plan_for(uc, pre) layers = plan.topological_order() # raises on cycles flat = [t.id for layer in layers for t in layer] assert len(flat) == len(plan.tasks), uc # ── Selection with complete bibliographic details ─────────── from src.pipeline.search_executor import ( # noqa: E402 render_selected_references, ) HITS = [ {"title": "AI-Powered Educational Agents", "doi": "10.3390/info16060469", "authors": ["Córdova-Esparza", "Terven", "Ramírez"], "year": 2025, "venue": "Information", "cited_by_count": 59, "sources": ["openalex"], "url": ""}, {"title": "Guidelines and Policies", "doi": "10.1234/hei.2025.7", "authors": ["Weber"], "year": 2025, "venue": "Higher Education Policy", "cited_by_count": 31, "sources": ["openalex", "s2"], "url": ""}, ] def test_selection_carries_authors_venue_and_doi(): """The selection contains full details, not only title and year.""" md = render_selected_references( {"selected": [{"doi": "10.3390/info16060469", "reason": "Passt."}]}, HITS) assert "Córdova-Esparza" in md assert "*Information*" in md assert "10.3390/info16060469" in md assert "59 citations" in md assert "→ Passt." in md def test_metadata_comes_from_the_api_not_the_model(): """Wrong details from the model must not come through.""" md = render_selected_references( {"selected": [{"doi": "10.3390/info16060469", "title": "Falscher Titel vom Modell", "authors": ["Erfundener Autor"], "year": 1999, "reason": "x"}]}, HITS) assert "Falscher Titel" not in md and "Erfundener Autor" not in md assert "AI-Powered Educational Agents" in md and "2025" in md def test_invented_dois_are_dropped_and_counted(): md = render_selected_references( {"selected": [{"doi": "10.3390/info16060469", "reason": "ok"}, {"doi": "10.9999/erfunden", "reason": "halluziniert"}]}, HITS) assert "10.9999" not in md assert "1 entry was discarded" in md def test_selection_falls_back_to_title_match(): """Without a DOI, but with an exact title, the hit stays usable.""" md = render_selected_references( {"selected": [{"title": "Guidelines and Policies", "reason": "ok"}]}, HITS) assert "Higher Education Policy" in md def test_gaps_and_next_round_are_kept(): md = render_selected_references( {"selected": [], "gaps": "Keine deutschsprachigen Quellen.", "next_round": "OECD durchsuchen."}, HITS) assert "Gaps in the hits" in md assert "Keine deutschsprachigen Quellen." in md assert "next search round" in md and "OECD durchsuchen." in md def test_empty_selection_is_stated_not_faked(): md = render_selected_references({}, HITS) assert "No selection made" in md def test_finder_plan_renders_the_selection_deterministically(): plan = _plan() ids = [t.id for t in plan.tasks] assert ids == ["STRATEGY", "QUERIES", "SEARCH", "ASSESS", "SELECTION"] assess = next(t for t in plan.tasks if t.id == "ASSESS") assert assess.output_format == "json" sel = next(t for t in plan.tasks if t.id == "SELECTION") assert sel.executor == "render_selection" # the JSON intermediate step must not end up in the report shown = [t.id for t in plan.tasks if t.show_in_report or not any(t.id in x.depends_on for x in plan.tasks)] assert "ASSESS" not in shown and "SELECTION" in shown