""" Tests for `src.pipeline.report_quality`. Acceptance: - check_report_quality returns a clean result for good/deficient reports - long report: segmented check - findings are merged across segments - hallucinations are marked as 'hallucinated' - on LLM failure: conservatively fulfilled=True, fallback_used=True - check_query_fulfillment works at the level of the user query - excerpt for a report that is too long (head + tail) """ import unittest from src.pipeline.models import ( ResearchPlan, ResearchQuestion, SourceExtract, ) from src.pipeline.report_quality import ( FULFILLMENT_REPORT_MAX, FulfillmentResult, QualityIssue, QUALITY_FULLTEXT_THRESHOLD, check_query_fulfillment, check_report_quality, dedupe_maengel, split_for_quality, ) from tests._helpers import MockLLM, async_test def _patch_adapter_with(llm: MockLLM): from src.llm import classifier_adapter class _Pass: def __init__(self, _dual): pass async def complete(self, messages, max_tokens=None): return await llm.complete(messages, max_tokens=max_tokens) original = classifier_adapter.HarvestModelAdapter classifier_adapter.HarvestModelAdapter = _Pass return original def _restore_adapter(original): from src.llm import classifier_adapter classifier_adapter.HarvestModelAdapter = original def _make_plan() -> ResearchPlan: return ResearchPlan( summary="Recherche zu DGX", questions=[ ResearchQuestion(id="F1", question="Was kostet die DGX?", search_terms=["x"], priority="hoch"), ResearchQuestion(id="F2", question="Stromverbrauch?", search_terms=["x"], priority="hoch"), ], ) # ─────────────────────────────────────────────────────────────────── # split_for_quality # ─────────────────────────────────────────────────────────────────── class TestSplitForQuality(unittest.TestCase): def test_empty_returns_empty(self): self.assertEqual(split_for_quality(""), []) def test_short_text_one_or_two_segments(self): # With little text and no headings: paragraph split text = "Kurzer Text mit zwei\n\nAbsätzen." segs = split_for_quality(text) self.assertGreaterEqual(len(segs), 1) # content is preserved self.assertEqual( "".join(segs).replace("\n\n", "").replace("\n", ""), text.replace("\n\n", "").replace("\n", ""), ) def test_heading_based_split_with_enough_headings(self): text = ( "## Einleitung\n\nText 1.\n\n" "## Hauptteil\n\nText 2.\n\n" "## Schluss\n\nText 3.\n\n" ) segs = split_for_quality(text) # 2-3 segments expected with this threshold self.assertGreaterEqual(len(segs), 2) # headings are contained in the segments all_text = "\n".join(segs) self.assertIn("Einleitung", all_text) self.assertIn("Hauptteil", all_text) self.assertIn("Schluss", all_text) # ─────────────────────────────────────────────────────────────────── # dedupe_maengel # ─────────────────────────────────────────────────────────────────── class TestDedupeMaengel(unittest.TestCase): def test_empty(self): self.assertEqual(dedupe_maengel([]), []) def test_unique_maengel_pass_through(self): issues = [ {"description": "M1", "kind": "missing", "segment": 1}, {"description": "M2", "kind": "redundant", "segment": 2}, ] out = dedupe_maengel(issues) self.assertEqual(len(out), 2) self.assertEqual(out[0].segments, [1]) self.assertEqual(out[1].segments, [2]) def test_duplicate_descriptions_merge(self): issues = [ {"description": "Gleicher Mangel", "kind": "missing", "segment": 1}, {"description": "Gleicher Mangel", "kind": "missing", "segment": 2}, {"description": "Gleicher Mangel", "kind": "missing", "segment": 3}, ] out = dedupe_maengel(issues) self.assertEqual(len(out), 1) self.assertEqual(out[0].segments, [1, 2, 3]) def test_case_insensitive_match(self): issues = [ {"description": "Halluzination bei Preisangabe", "segment": 1}, {"description": "HALLUZINATION BEI PREISANGABE", "segment": 2}, ] out = dedupe_maengel(issues) self.assertEqual(len(out), 1) self.assertEqual(sorted(out[0].segments), [1, 2]) def test_string_maengel_kept(self): issues = ["Roh-Strings ohne Dict-Struktur"] out = dedupe_maengel(issues) self.assertEqual(len(out), 1) self.assertEqual(out[0].description, "Roh-Strings ohne Dict-Struktur") def test_empty_descriptions_skipped(self): out = dedupe_maengel([ {"description": "", "kind": "missing"}, {"description": " ", "kind": "missing"}, ]) self.assertEqual(out, []) # ─────────────────────────────────────────────────────────────────── # check_report_quality — full text # ─────────────────────────────────────────────────────────────────── class TestCheckReportQualityFulltext(unittest.TestCase): @async_test async def test_clean_report_passes(self): llm = MockLLM().queue_json({ "passed": True, "rating": "good", "issues": [], }) original = _patch_adapter_with(llm) try: result = await check_report_quality( report="## Bericht\n\nKurz aber gut.", extracts=[ SourceExtract("u", "t", "F1", "Fakt", polarity="positive"), ], plan=_make_plan(), dual_llm=None, ) finally: _restore_adapter(original) self.assertTrue(result.passed) self.assertEqual(result.rating, "good") self.assertEqual(result.issues, []) self.assertEqual(result.segments, 1) self.assertFalse(result.fallback_used) @async_test async def test_hallucination_detected(self): llm = MockLLM().queue_json({ "passed": False, "rating": "poor", "issues": [{ "question_id": "F1", "kind": "hallucinated", "description": "Preis 500$ steht nicht in den Extrakten", "suggestion": "Mit Quelle belegen oder weglassen", }], }) original = _patch_adapter_with(llm) try: result = await check_report_quality( report="## Bericht\n\nDie DGX kostet 500$.", extracts=[ SourceExtract("u", "t", "F1", "Andere Aussage", polarity="positive"), ], plan=_make_plan(), dual_llm=None, ) finally: _restore_adapter(original) self.assertFalse(result.passed) self.assertEqual(len(result.issues), 1) self.assertEqual(result.issues[0].kind, "hallucinated") @async_test async def test_empty_report_returns_fallback(self): result = await check_report_quality( report="", extracts=[], plan=None, dual_llm=None, ) self.assertTrue(result.passed) self.assertTrue(result.fallback_used) @async_test async def test_llm_failure_returns_fallback(self): from src.llm import classifier_adapter class _Broken: def __init__(self, _dual): pass async def complete(self, messages, max_tokens=None): raise RuntimeError("LLM down") original = classifier_adapter.HarvestModelAdapter classifier_adapter.HarvestModelAdapter = _Broken try: result = await check_report_quality( report="## Bericht", extracts=[ SourceExtract("u", "t", "F1", "x", polarity="positive"), ], plan=_make_plan(), dual_llm=None, ) finally: classifier_adapter.HarvestModelAdapter = original self.assertTrue(result.passed) self.assertTrue(result.fallback_used) @async_test async def test_only_positive_extracts_in_prompt(self): """Negative/meta extracts do NOT go into the prompt as the factual basis.""" llm = MockLLM().queue_json({"passed": True, "issues": []}) captured = {} from src.llm import classifier_adapter class _Capturing: def __init__(self, _dual): pass async def complete(self, messages, max_tokens=None): captured["messages"] = messages return await llm.complete(messages, max_tokens=max_tokens) original = classifier_adapter.HarvestModelAdapter classifier_adapter.HarvestModelAdapter = _Capturing try: await check_report_quality( report="## Bericht", extracts=[ SourceExtract("u", "t", "F1", "POSITIV-FAKT", polarity="positive"), SourceExtract("u", "t", "F1", "NEGATIV-FAKT", polarity="negative"), SourceExtract("u", "t", "F1", "META-FAKT", polarity="meta"), ], plan=_make_plan(), dual_llm=None, ) finally: classifier_adapter.HarvestModelAdapter = original prompt_text = "\n".join( m.get("content", "") for m in captured["messages"] ) self.assertIn("POSITIV-FAKT", prompt_text) self.assertNotIn("NEGATIV-FAKT", prompt_text) self.assertNotIn("META-FAKT", prompt_text) # ─────────────────────────────────────────────────────────────────── # check_report_quality — segmented # ─────────────────────────────────────────────────────────────────── class TestCheckReportQualitySegmented(unittest.TestCase): @async_test async def test_long_report_uses_segmented_path(self): # report > QUALITY_FULLTEXT_THRESHOLD with headings long_report = ( "## Einleitung\n\n" + "Text. " * 200 + "## Hauptteil\n\n" + "Text. " * 800 + "## Schluss\n\n" + "Text. " * 1500 ) # >15k characters required self.assertGreater(len(long_report), QUALITY_FULLTEXT_THRESHOLD) # three LLM answers (one per segment) llm = MockLLM() for _ in range(3): llm.queue_json({ "passed": True, "rating": "good", "issues": [], }) original = _patch_adapter_with(llm) try: result = await check_report_quality( report=long_report, extracts=[ SourceExtract("u", "t", "F1", "x", polarity="positive"), ], plan=_make_plan(), dual_llm=None, ) finally: _restore_adapter(original) # more than 1 segment checked self.assertGreaterEqual(result.segments, 2) self.assertTrue(result.passed) @async_test async def test_segments_merge_duplicate_maengel(self): """If two segments report the same finding, it is merged.""" long_report = ( "## A\n\n" + "Text. " * 1500 + "## B\n\n" + "Text. " * 1500 + "## C\n\n" + "Text. " * 1500 ) self.assertGreater(len(long_report), QUALITY_FULLTEXT_THRESHOLD) # both segments report the same finding llm = MockLLM() for _ in range(3): llm.queue_json({ "passed": False, "rating": "poor", "issues": [{ "question_id": "F1", "kind": "redundant", "description": "Selber Punkt mehrfach erwähnt", "suggestion": "Eine Stelle löschen", }], }) original = _patch_adapter_with(llm) try: result = await check_report_quality( report=long_report, extracts=[ SourceExtract("u", "t", "F1", "x", polarity="positive"), ], plan=_make_plan(), dual_llm=None, ) finally: _restore_adapter(original) # findings were merged — only 1 entry self.assertEqual(len(result.issues), 1) self.assertEqual(result.issues[0].kind, "redundant") # but the segment field shows that several segments are affected self.assertGreaterEqual(len(result.issues[0].segments), 2) # ─────────────────────────────────────────────────────────────────── # check_query_fulfillment # ─────────────────────────────────────────────────────────────────── class TestCheckQueryFulfillment(unittest.TestCase): @async_test async def test_fulfilled_returns_true(self): llm = MockLLM().queue_json({ "fulfilled": True, "assessment": "Anfrage komplett beantwortet", "rework": "", }) original = _patch_adapter_with(llm) try: result = await check_query_fulfillment( query="Was kostet die DGX?", report="## DGX-Preis\n\nDie DGX kostet $415k.", dual_llm=None, ) finally: _restore_adapter(original) self.assertTrue(result.fulfilled) self.assertIn("beantwortet", result.assessment.lower()) @async_test async def test_not_fulfilled_with_nacharbeit(self): llm = MockLLM().queue_json({ "fulfilled": False, "assessment": "Nur Preis genannt, Stromverbrauch fehlt", "rework": "Ergänze einen Abschnitt zum TDP", }) original = _patch_adapter_with(llm) try: result = await check_query_fulfillment( query="Preis und Stromverbrauch der DGX?", report="## Preis\n\n$415k.", dual_llm=None, ) finally: _restore_adapter(original) self.assertFalse(result.fulfilled) self.assertIn("TDP", result.rework) @async_test async def test_empty_query_returns_fallback(self): result = await check_query_fulfillment("", "Bericht", dual_llm=None) self.assertTrue(result.fulfilled) self.assertTrue(result.fallback_used) @async_test async def test_long_report_uses_excerpt(self): """For a very long report: head + [...] + tail.""" long_report = "X" * (FULFILLMENT_REPORT_MAX * 2) captured = {} from src.llm import classifier_adapter from tests._helpers import MockLLM as _MockLLM sub = _MockLLM().queue_json({"fulfilled": True, "assessment": "ok", "rework": ""}) class _Capturing: def __init__(self, _dual): pass async def complete(self, messages, max_tokens=None): captured["messages"] = messages return await sub.complete(messages, max_tokens=max_tokens) original = classifier_adapter.HarvestModelAdapter classifier_adapter.HarvestModelAdapter = _Capturing try: await check_query_fulfillment( query="Test", report=long_report, dual_llm=None, ) finally: classifier_adapter.HarvestModelAdapter = original prompt = "\n".join( m.get("content", "") for m in captured["messages"] ) # excerpt marker self.assertIn("[...]", prompt) # the prompt itself is smaller than the original text self.assertLess(len(prompt), len(long_report)) @async_test async def test_llm_failure_returns_fallback(self): from src.llm import classifier_adapter class _Broken: def __init__(self, _dual): pass async def complete(self, messages, max_tokens=None): raise RuntimeError("oops") original = classifier_adapter.HarvestModelAdapter classifier_adapter.HarvestModelAdapter = _Broken try: result = await check_query_fulfillment( query="?", report="Bericht", dual_llm=None, ) finally: classifier_adapter.HarvestModelAdapter = original self.assertTrue(result.fulfilled) self.assertTrue(result.fallback_used) if __name__ == "__main__": unittest.main() def test_german_answers_are_mapped_to_the_english_contract(): """A model answering with the former German keys/values still parses.""" from src.pipeline.report_quality import ( _english_contract, ) q = _english_contract({"bestanden": False, "gesamtbewertung": "mangelhaft", "maengel": [{"art": "halluziniert", "beschreibung": "x"}]}) assert q == {"passed": False, "rating": "poor", "issues": [{"kind": "hallucinated", "description": "x"}]} assert QualityIssue.from_dict(q["issues"][0]).kind == "hallucinated" f = FulfillmentResult.from_dict(_english_contract({"erfuellt": False, "nacharbeit": "y"})) assert f.fulfilled is False and f.rework == "y"