backend/tests/test_e2e_spec.py

from __future__ import annotations

import io
import zipfile
from pathlib import Path
from unittest.mock import patch

import pytest
from docx import Document
from fastapi.testclient import TestClient

from backend.config import settings
from backend.core import ingest, section_mapper, template_discoverer
from backend.core.report_assembler import to_docx, to_preview
from backend.core.rag_store import TIER_MASTER, TIER_REFERENCE, get_rag_store
from backend.models.schema import RatingSystem, RatingValue


def _ingest_bundle(tenant: str, tpl: Path, paras: Path) -> None:
    ingest.ingest_report_template(tenant, tpl)
    ingest.ingest_standard_paragraphs(tenant, paras)
    ingest.ingest_reference(tenant, paras)


def _write_canonical_bundle(tmp_path: Path) -> tuple[Path, Path]:
    tpl = tmp_path / "template.docx"
    doc = Document()
    doc.add_paragraph("RICS Home Survey Level 3")
    doc.save(str(tpl))

    paras = tmp_path / "paragraphs.docx"
    doc2 = Document()
    doc2.add_paragraph("Section D1 - Chimney stacks")
    doc2.add_paragraph("The chimney stacks were inspected from ground level.")
    doc2.add_paragraph("Section D2 - Roof coverings")
    doc2.add_paragraph("The roof covering comprises slate laid to pitched rafters.")
    doc2.save(str(paras))
    return tpl, paras


def _docx_coloured_runs(data: bytes) -> list[str]:
    """Return 6-char hex colour values from DOCX run properties."""
    colours: list[str] = []
    with zipfile.ZipFile(io.BytesIO(data)) as zf:
        xml = zf.read("word/document.xml").decode("utf-8")
    for fragment in xml.split("<w:color"):
        if 'w:val="' not in fragment:
            continue
        start = fragment.find('w:val="') + len('w:val="')
        end = fragment.find('"', start)
        val = fragment[start:end]
        if len(val) == 6 and all(c in "0123456789ABCDEFabcdef" for c in val):
            colours.append(val.upper())
    return colours


def _docx_bold_runs(data: bytes) -> int:
    with zipfile.ZipFile(io.BytesIO(data)) as zf:
        xml = zf.read("word/document.xml").decode("utf-8")
    return xml.count("<w:b/>") + xml.count("<w:b ")


def test_e2e_rating_docx_only_on_has_rating_field(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "data_dir", str(tmp_path / "data"))
    tpl, paras = _write_canonical_bundle(tmp_path)
    tenant = "t-rated"
    _ingest_bundle(tenant, tpl, paras)

    schema = template_discoverer.load_schema(tenant)
    schema.rating_system = RatingSystem(
        detected=True,
        name="Defect Grade",
        format_template="Defect Grade [VALUE]",
        values=[
            RatingValue(value="1", colour="00B050"),
            RatingValue(value="2", colour="ED7D31"),
        ],
    )
    for sec in schema.sections:
        sec.has_rating_field = sec.id == "D1"
    template_discoverer.save_schema(tenant, schema)

    result = section_mapper.generate_report(
        tenant,
        "D1: mortar open joints defect grade 2\n\nD2: slate slipped to valley",
        property_type="house",
        tenure="freehold",
        interference_level="maximum",
    )
    for sec in result.sections:
        if sec.section_id == "D1":
            sec.rating_value = "2"
        elif sec.section_id == "D3":
            sec.rating_value = None

    docx = to_docx(result, schema)
    colours = _docx_coloured_runs(docx)
    assert colours, "expected rating colour on rated section"
    assert _docx_bold_runs(docx) >= 1


def test_e2e_no_rating_zero_annotations(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "data_dir", str(tmp_path / "data"))
    tpl, paras = _write_canonical_bundle(tmp_path)
    tenant = "t-norating"
    _ingest_bundle(tenant, tpl, paras)

    schema = template_discoverer.load_schema(tenant)
    schema.rating_system = RatingSystem(detected=False)
    template_discoverer.save_schema(tenant, schema)

    result = section_mapper.generate_report(
        tenant,
        "D1: mortar open joints rating 2\n\nD2: slate slipped",
        property_type="house",
        tenure="freehold",
        interference_level="maximum",
    )
    docx = to_docx(result, schema)
    assert not _docx_coloured_runs(docx)
    assert b"Defect Grade" not in docx


def test_e2e_new_firm_onboarding_fixture(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "data_dir", str(tmp_path / "data"))
    tpl, paras = _write_canonical_bundle(tmp_path)
    tenant = "new-firm"
    _ingest_bundle(tenant, tpl, paras)

    schema = template_discoverer.load_schema(tenant)
    assert schema.additional_metadata.get("parent_section_count") == 14
    assert "D1" in schema.section_ids()

    result = section_mapper.generate_report(
        tenant, "chimney stack brick spalling noted", interference_level="maximum"
    )
    ids = {s.section_id for s in result.sections if s.status != "empty"}
    assert "D1" in ids


def test_e2e_grounding_adversarial_unmatched(monkeypatch, tmp_path):
    monkeypatch.setattr(settings, "data_dir", str(tmp_path / "data"))
    monkeypatch.setattr(settings, "openai_api_key", "sk-test")

    tpl, paras = _write_canonical_bundle(tmp_path)
    tenant = "t-adv"
    _ingest_bundle(tenant, tpl, paras)

    invented = (
        "The chimney was inspected.\n"
        "[UNMATCHED_OBSERVATION: completely invented asbestos contamination]"
    )

    with patch("backend.core.reference_mapper.openai_client.is_available", return_value=True), patch(
        "backend.core.reference_mapper.openai_client.chat_text",
        return_value=invented,
    ):
        result = section_mapper.generate_report(
            tenant, "D1: chimney hairline crack", interference_level="maximum"
        )

    d1 = next(s for s in result.sections if s.section_id == "D1")
    assert "asbestos" not in d1.text.lower()


def test_e2e_pii_adversarial_name_and_postcode(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "data_dir", str(tmp_path / "data"))
    tpl, paras = _write_canonical_bundle(tmp_path)
    tenant = "t-pii-adv"
    _ingest_bundle(tenant, tpl, paras)

    notes = "D1: crack near stack\nClient Jane Smith at SW1A 1AA"
    result = section_mapper.generate_report(
        tenant, notes, property_type="flat", tenure="leasehold", interference_level="maximum"
    )
    schema = template_discoverer.load_schema(tenant)
    docx = to_docx(result, schema).decode("latin-1", errors="ignore").lower()
    assert "sw1a" not in docx
    assert "jane smith" not in docx


def test_empty_faiss_returns_400(client, configured_master):
    get_rag_store().clear_tier(settings.default_tenant_id, TIER_REFERENCE)
    r = client.post(
        "/auth/register",
        json={"tenant_id": settings.default_tenant_id, "password": "secret123"},
    )
    token = r.json()["access_token"]
    r = client.post(
        "/api/report/generate",
        json={
            "raw_notes": "roof slate slipped",
            "property_type": "house",
            "tenure": "freehold",
        },
        headers={"Authorization": f"Bearer {token}"},
    )
    assert r.status_code == 400
    assert "ingested" in r.json()["detail"].lower() or "past reports" in r.json()["detail"].lower()


def test_grounding_alert_threshold(client, configured_master, monkeypatch):
    monkeypatch.setattr(settings, "grounding_alert_threshold", 0.0)
    r = client.post(
        "/auth/register",
        json={"tenant_id": settings.default_tenant_id, "password": "secret456"},
    )
    token = r.json()["access_token"]
    r = client.post(
        "/api/report/preview",
        json={
            "raw_notes": "chimney cracked\n\nroof slate",
            "property_type": "house",
            "tenure": "freehold",
        },
        headers={"Authorization": f"Bearer {token}"},
    )
    assert r.status_code == 200
    body = r.json()
    assert "manual_review_required" in body
    assert "review_ratio" in body


def test_reference_upload_http(client, configured_master, tmp_path):
    ref = tmp_path / "past.docx"
    doc = Document()
    doc.add_paragraph("Past report boilerplate about roof coverings at a generic property.")
    doc.save(str(ref))

    r = client.post(
        "/auth/register",
        json={"tenant_id": settings.default_tenant_id, "password": "secret789"},
    )
    token = r.json()["access_token"]
    with ref.open("rb") as fh:
        r = client.post(
            "/api/upload/reference",
            files={"file": ("past.docx", fh, "application/vnd.openxmlformats-officedocument.wordprocessingml.document")},
            headers={"Authorization": f"Bearer {token}"},
        )
    assert r.status_code == 201
    assert r.json()["tier"] == "reference"
    assert r.json()["ingested_chunks"] >= 1


def test_schema_roundtrip_additional_metadata(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "data_dir", str(tmp_path / "data"))
    schema = template_discoverer.install_canonical_schema("t-meta")
    schema.additional_metadata = {**schema.additional_metadata, "firm": "Acme Surveys"}
    template_discoverer.save_schema("t-meta", schema)
    loaded = template_discoverer.load_schema("t-meta")
    assert loaded is not None
    assert loaded.tenant_id == "t-meta"
    assert loaded.additional_metadata["firm"] == "Acme Surveys"
    assert loaded.additional_metadata.get("parent_section_count") == 14