"""Tests for the two-layer hallucination / non-invention postprocessor. Layer 1 (regex) tests run without any OpenAI key. Layer 2 (LLM grounding) tests mock the OpenAI call. """ from __future__ import annotations import json from unittest.mock import AsyncMock, MagicMock, patch import pytest from app.generator.postprocess import ( GroundingViolation, _apply_violations, _llm_grounding_check, async_enforce_verify, enforce_verify, ) # --------------------------------------------------------------------------- # Layer 1 — synchronous regex pass # --------------------------------------------------------------------------- class TestEnforceVerifyRegex: """enforce_verify (regex-only, no LLM) tests.""" def test_verified_number_passes(self) -> None: text = "The property has 3 bedrooms." result = enforce_verify(text=text, bullets=["3 bedroom semi-detached"], snippets=[]) assert "3" in result def test_unverified_number_replaced(self) -> None: text = "The property has 4 bedrooms." result = enforce_verify(text=text, bullets=["3 bedroom semi-detached"], snippets=[]) assert "4" not in result assert result == "The property has ." def test_verified_postcode_passes(self) -> None: text = "The property is located at SW1A 1AA." result = enforce_verify(text=text, bullets=["SW1A 1AA"], snippets=[]) assert "SW1A 1AA" in result def test_unverified_postcode_replaced(self) -> None: text = "The property is located at SW1A 1AA." result = enforce_verify(text=text, bullets=["2 bedroom flat"], snippets=[]) assert "SW1A 1AA" not in result assert result == "The property is located at ." def test_allowlisted_entity_never_replaced(self) -> None: """Standard RICS terms like 'Ground Floor' must never be flagged.""" text = "The Ground Floor shows signs of wear." result = enforce_verify(text=text, bullets=["slight wear noted"], snippets=[]) assert "Ground Floor" in result def test_allowlisted_entity_cavity_wall(self) -> None: text = "Cavity Wall construction observed throughout." result = enforce_verify(text=text, bullets=["brick construction"], snippets=[]) assert "Cavity Wall" in result def test_allowlisted_condition_rating(self) -> None: text = "Condition Rating 2 is assigned." result = enforce_verify(text=text, bullets=["minor defect noted"], snippets=[]) # "Condition Rating" is allowlisted; "2" must appear in source to survive assert "Condition Rating" in result def test_unverified_named_entity_replaced(self) -> None: text = "Surveyed by Smith Associates Ltd." result = enforce_verify( text=text, bullets=["inspection carried out"], snippets=[] ) # "Smith Associates" is a made-up firm not in the source assert "Smith Associates" not in result def test_negated_context_number_in_snippet(self) -> None: """A number present in snippets should not be replaced.""" text = "Wall thickness is approximately 275mm." result = enforce_verify( text=text, bullets=[], snippets=["solid brick walls 275mm DPC visible"], ) assert "275" in result def test_empty_text_passes(self) -> None: result = enforce_verify(text="", bullets=[], snippets=[]) assert result == "" def test_legacy_verify_tags_removed(self) -> None: text = "The roof is [VERIFY: 5 years old]." result = enforce_verify(text=text, bullets=[], snippets=[]) assert "[VERIFY:" not in result assert result == "The roof is ." # --------------------------------------------------------------------------- # Layer 2 — LLM grounding (_llm_grounding_check) # --------------------------------------------------------------------------- def _make_openai_response(violations: list[dict], score: float = 1.0) -> MagicMock: """Build a mock OpenAI response object.""" content = json.dumps({"violations": violations, "grounding_score": score}) msg = MagicMock() msg.content = content choice = MagicMock() choice.message = msg resp = MagicMock() resp.choices = [choice] return resp @pytest.mark.asyncio async def test_llm_grounding_no_violations() -> None: """When the LLM returns no violations, grounding_score=1.0 and text is unchanged.""" mock_resp = _make_openai_response(violations=[], score=1.0) with patch( "app.llm.openai_chat.chat_completions_create", new_callable=AsyncMock, ) as mock_chat: mock_chat.return_value = json.dumps({"violations": [], "grounding_score": 1.0}) result = await _llm_grounding_check( text="The roof appeared in fair condition.", bullets=["roof fair condition"], snippets=[], openai_api_key="sk-test", ) assert result.violations == [] assert result.grounding_score == 1.0 assert result.method == "llm" @pytest.mark.asyncio async def test_llm_grounding_violation_returned() -> None: """When the LLM flags a claim, it appears in violations.""" payload = { "violations": [ { "original": "4 bedrooms", "replacement": "Information not provided in source document.", "reason": "bedroom count not in source", } ], "grounding_score": 0.7, } with patch( "app.llm.openai_chat.chat_completions_create", new_callable=AsyncMock, ) as mock_chat: mock_chat.return_value = json.dumps(payload) result = await _llm_grounding_check( text="The property has 4 bedrooms.", bullets=["3 bed semi"], snippets=[], openai_api_key="sk-test", ) assert len(result.violations) == 1 assert result.violations[0].original == "4 bedrooms" assert result.grounding_score == 0.7 @pytest.mark.asyncio async def test_llm_grounding_falls_back_on_openai_error() -> None: """When the OpenAI call raises, we fall back gracefully (no violations, score=1.0).""" with patch( "app.llm.openai_chat.chat_completions_create", new_callable=AsyncMock, ) as mock_chat: mock_chat.side_effect = RuntimeError("network error") result = await _llm_grounding_check( text="Some text.", bullets=["bullet"], snippets=[], openai_api_key="sk-test", ) assert result.violations == [] assert result.method == "regex_fallback" @pytest.mark.asyncio async def test_llm_grounding_falls_back_on_bad_json() -> None: """Malformed JSON from the LLM must not crash the pipeline.""" with patch( "app.llm.openai_chat.chat_completions_create", new_callable=AsyncMock, ) as mock_chat: mock_chat.return_value = "not-json" result = await _llm_grounding_check( text="Some text.", bullets=["bullet"], snippets=[], openai_api_key="sk-test", ) assert result.violations == [] assert result.method == "regex_fallback" # --------------------------------------------------------------------------- # _apply_violations # --------------------------------------------------------------------------- def test_apply_violations_replaces_exact_match() -> None: text = "The property has 4 bedrooms." violations = [ GroundingViolation( original="4 bedrooms", replacement="Information not provided in source document.", reason="not in source", ) ] result = _apply_violations(text, violations) assert "4 bedrooms" not in result assert "Information not provided" in result def test_apply_violations_skips_missing_original() -> None: """If the original phrase isn't in the text (e.g. regex already replaced it), skip.""" text = "The property has 3 bedrooms." violations = [ GroundingViolation( original="4 bedrooms", replacement="Information not provided in source document.", reason="not in source", ) ] result = _apply_violations(text, violations) assert result == text # unchanged def test_apply_violations_empty_original_skipped() -> None: violations = [GroundingViolation(original="", replacement="X", reason="")] result = _apply_violations("Some text.", violations) assert result == "Some text." # --------------------------------------------------------------------------- # async_enforce_verify — full two-layer integration # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_async_enforce_verify_no_key_uses_regex_only() -> None: """Without an API key, only the regex pass runs — no LLM call.""" text = "The property has 4 bedrooms." with patch("app.generator.postprocess._llm_grounding_check") as mock_llm: result = await async_enforce_verify( text=text, bullets=["3 bedroom semi"], snippets=[], openai_api_key="", # no key → skip LLM ) mock_llm.assert_not_called() assert "4" not in result # regex still catches it @pytest.mark.asyncio async def test_async_enforce_verify_with_key_calls_llm() -> None: """With an API key, the LLM grounding pass is called after the regex pass.""" text = "The roof appeared in fair condition." with patch( "app.generator.postprocess._llm_grounding_check", new_callable=AsyncMock ) as mock_llm: from app.generator.postprocess import GroundingResult mock_llm.return_value = GroundingResult( violations=[], grounding_score=1.0, method="llm" ) result = await async_enforce_verify( text=text, bullets=["roof fair condition"], snippets=[], openai_api_key="sk-test", ) mock_llm.assert_called_once() assert "fair condition" in result @pytest.mark.asyncio async def test_async_enforce_verify_llm_violation_applied() -> None: """LLM violations are applied on top of the regex-clean text.""" text = "The property has 3 bedrooms and Smith Associates signed off the survey." from app.generator.postprocess import GroundingResult with patch( "app.generator.postprocess._llm_grounding_check", new_callable=AsyncMock ) as mock_llm: mock_llm.return_value = GroundingResult( violations=[ GroundingViolation( original="Smith Associates signed off the survey", replacement="Information not provided in source document.", reason="firm name not in source", ) ], grounding_score=0.8, method="llm", ) result = await async_enforce_verify( text=text, bullets=["3 bedroom semi"], snippets=[], openai_api_key="sk-test", ) assert "Smith Associates" not in result assert "Information not provided" not in result assert result == "The property has 3 bedrooms and signed off the survey." assert "3" in result # verified number survives # --------------------------------------------------------------------------- # Level-1 advice sanitiser # --------------------------------------------------------------------------- # # RICS Level 1 (Condition Report) is observation-only. The system prompt # forbids advice phrasing, but the LLM occasionally leaks "we recommend ..." # / "should be replaced" — language that turns the L1 product into a partial # L2. The agentic inspector loop has no retry loop (unlike the legacy LCEL # path) so a deterministic regex sanitiser is the production safety net. class TestL1AdviceSanitiser: """`strip_l1_advice` — observation-only enforcement for Level 1.""" def test_strips_we_recommend_sentence(self) -> None: from app.generator.postprocess import strip_l1_advice text = ( "The roof covering is concrete tile and broadly weathertight. " "We recommend specialist flat roof investigation. " "No active leakage was observed at the time of inspection." ) cleaned = strip_l1_advice(text) assert "we recommend" not in cleaned.lower() # Surrounding observation prose is preserved. assert "concrete tile" in cleaned assert "no active leakage" in cleaned.lower() def test_strips_should_be_replaced(self) -> None: from app.generator.postprocess import strip_l1_advice text = ( "The boiler is a 2014 condensing unit. " "It should be replaced before purchase. " "Service records were not made available." ) cleaned = strip_l1_advice(text) assert "should be replaced" not in cleaned.lower() assert "boiler" in cleaned.lower() assert "service records" in cleaned.lower() def test_strips_obtain_specialist_report(self) -> None: from app.generator.postprocess import strip_l1_advice text = ( "Visible cracks were noted at the rear elevation. " "Obtain a structural engineer's report before proceeding." ) cleaned = strip_l1_advice(text) # The advisory sentence is dropped; the observation remains. assert "structural engineer" not in cleaned.lower() assert "visible cracks" in cleaned.lower() def test_pure_observation_passes_unchanged(self) -> None: from app.generator.postprocess import strip_l1_advice text = ( "The kitchen units appeared in serviceable condition. " "Surface scratches were noted on three drawer fronts." ) cleaned = strip_l1_advice(text) assert cleaned.strip() == text.strip() def test_mid_text_advice_removal_preserves_sentence_separator(self) -> None: """Regression: the non-greedy `[^.!?]*?` prefix in `_L1_ADVICE_SENTENCE_RE` consumes the leading whitespace before the advice marker (to position at `\\b`), and the trailing `\\s*` consumes the whitespace after the sentence terminator. Replacing the match with `""` therefore collapsed adjacent sentences into one, producing output like ``"The walls are sound.The floors are level."`` — no space between the surrounding sentences. The fix replaces with a single space so the existing `\\s{2,}` collapse + `.strip()` keep exactly one space between any two surviving sentences.""" from app.generator.postprocess import strip_l1_advice text = ( "The walls are sound. We recommend repairs. The floors are level." ) cleaned = strip_l1_advice(text) # Both surviving sentences must be present AND separated by exactly # one space — never glued together. assert "sound. The" in cleaned assert "sound.The" not in cleaned, ( "Regression: mid-text advice removal collapsed adjacent sentences" ) def test_consecutive_advice_sentences_collapse_cleanly(self) -> None: """Two back-to-back advice sentences must both be removed and the surrounding observations joined with a single space (no glue, no double space).""" from app.generator.postprocess import strip_l1_advice text = ( "The walls are sound. We recommend X. We recommend Y. " "The floors are level." ) cleaned = strip_l1_advice(text) assert "recommend" not in cleaned.lower() assert "sound. The floors" in cleaned # No double space (the \\s{2,} collapse must catch consecutive-strip # boundary effects). assert " " not in cleaned def test_advice_at_text_start_strips_without_leading_space(self) -> None: """Advice at the very start of the input must be removed and the surviving text must not begin with a whitespace gap.""" from app.generator.postprocess import strip_l1_advice text = "We recommend a full survey. The walls are sound." cleaned = strip_l1_advice(text) # The remaining observation is the only content; it must not start # with the placeholder space the regex inserts at the strip boundary. assert cleaned == "The walls are sound." def test_advice_at_text_end_strips_without_trailing_space(self) -> None: from app.generator.postprocess import strip_l1_advice text = "The walls are sound. We recommend a full survey." cleaned = strip_l1_advice(text) assert cleaned == "The walls are sound." def test_all_advice_input_returns_l1_placeholder(self) -> None: """When every sentence is advisory, return an L1-appropriate placeholder rather than emit an empty field. The agentic pipeline writes this into ``recommendations`` for L1 sections so the renderer never shows an empty heading. """ from app.generator.postprocess import strip_l1_advice text = "We recommend a full electrical inspection. Obtain a specialist report." cleaned = strip_l1_advice(text) assert "recommend" not in cleaned.lower() assert "level 1" in cleaned.lower() assert "observation only" in cleaned.lower() def test_payload_helper_sweeps_all_five_fields(self) -> None: """`strip_l1_advice_payload` must touch every string field of the agentic submit payload — otherwise advice could leak through any one of the five user-visible fields the report renderer reads. """ from app.generator.postprocess import strip_l1_advice_payload payload = { "executive_summary": "Condition note. We recommend further investigation.", "property_description": "A two-bedroom flat on the first floor.", "condition_assessment": "Damp staining noted. Should be replaced soon.", "defects_and_risks": "Crack at rear. You should obtain a quotation.", "recommendations": "We recommend specialist review of the consumer unit.", } cleaned = strip_l1_advice_payload(payload) for field in ( "executive_summary", "condition_assessment", "defects_and_risks", "recommendations", ): assert "recommend" not in cleaned[field].lower() assert "should be replaced" not in cleaned[field].lower() assert "you should" not in cleaned[field].lower() # Pure-observation field passes through untouched. assert cleaned["property_description"].strip() == payload["property_description"].strip() def test_grounding_system_prompts_for_contradiction_first() -> None: """The user reported "robust steel frame" output when the source said "cavity brick wall". That's a CONTRADICTION (source explicitly disagrees), not a missing fact. The previous grounding prompt told the auditor to "be CONSERVATIVE — better to allow than over-flag", which let exactly these contradictions through. Lock in the new contradiction-first framing so a future edit can't silently regress to lenient mode. """ from app.generator.postprocess import _GROUNDING_SYSTEM lower = _GROUNDING_SYSTEM.lower() # Tier 1 (contradictions) must be present and explicit. assert "contradiction" in lower assert "tier 1" in lower # Worked-example contrasts the user explicitly cited. assert "steel frame" in lower or "single-glazed" in lower # The old "be conservative" framing is gone; new framing prefers flagging. assert "be conservative" not in lower assert "false negative" in lower or "prefer flagging" in lower