app/tests/test_postprocess.py

"""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