smoke_test.py

"""End-to-end smoke test: upload → ingest → create report → generate → read sections."""

import asyncio
import json

from io import BytesIO

import httpx
from docx import Document as DocxDoc


async def run() -> None:
    headers = {"X-Tenant-ID": "demo_tenant"}
    async with httpx.AsyncClient(
        base_url="http://localhost:8000", headers=headers, timeout=60.0
    ) as c:

        # ── Step 1: Upload a DOCX ─────────────────────────────────────────────
        buf = BytesIO()
        d = DocxDoc()
        d.add_heading("SAMPLE: 12 High Street, London NW3", 1)
        d.add_paragraph("SAMPLE: The property is a semi-detached house built circa 1978.")
        d.add_paragraph("SAMPLE: Floor area approximately 95 sqm. Located in NW3 postcode.")
        d.add_paragraph("SAMPLE: The roof is covered with concrete interlocking tiles in fair condition.")
        d.add_paragraph("SAMPLE: External walls are cavity brick construction with UPVC double-glazed windows.")
        d.save(buf)
        buf.seek(0)

        r = await c.post(
            "/upload",
            files={"file": ("sample.docx", buf, "application/vnd.openxmlformats-officedocument.wordprocessingml.document")},
            data={"tenant_id": "demo_tenant"},
        )
        up = r.json()
        print(f"1. Upload: {r.status_code} | doc_id: {up['document_id']}")
        doc_id = up["document_id"]

        # ── Step 2: Poll until ingested ───────────────────────────────────────
        for _ in range(20):
            await asyncio.sleep(2)
            r = await c.get(f"/documents/{doc_id}/status")
            st = r.json()["status"]
            print(f"   Ingest status: {st}")
            if st in ("complete", "failed"):
                break

        # ── Step 3: Create a report ───────────────────────────────────────────
        r = await c.post(f"/reports?document_id={doc_id}")
        rep = r.json()
        print(f"3. Create report: {r.status_code} | report_id: {rep.get('report_id')}")
        report_id = rep.get("report_id")

        # ── Step 4: Generate section E2 (Roof coverings) ─────────────────────
        payload = {
            "template_id": "E2",
            "bullets": [
                "Roof: concrete interlocking tiles, fair condition",
                "Hip roof with valley gutter, no significant defects observed",
                "One cracked tile to south slope, recommend monitor",
            ],
        }
        r = await c.post(f"/reports/{report_id}/generate", json=payload)
        print(f"4. Generate: {r.status_code} | {r.json()}")

        # ── Step 5: Poll until complete ───────────────────────────────────────
        for _ in range(20):
            await asyncio.sleep(2)
            r = await c.get(f"/reports/{report_id}/status")
            st = r.json()["status"]
            print(f"   Report status: {st}")
            if st in ("complete", "failed"):
                break

        # ── Step 6: Get generated sections ────────────────────────────────────
        r = await c.get(f"/reports/{report_id}/sections")
        data = r.json()
        print(f"6. Sections: {r.status_code}")
        for code, sec in data.get("sections", {}).items():
            print(f"\n   [{code}] confidence={sec['confidence']:.2f}  cached={sec['cached']}")
            print(f"   TEXT: {sec['text'][:300]}")
            if sec.get("provenance"):
                print(f"   PROVENANCE: {json.dumps(sec['provenance'][:2], indent=2)}")

        # ── Step 7: Generate the same section again — should be cached ─────────
        print("\n7. Generating same section again (should use cache)…")
        r = await c.post(f"/reports/{report_id}/generate", json=payload)
        await asyncio.sleep(3)
        r = await c.get(f"/reports/{report_id}/sections")
        data2 = r.json()
        for code, sec in data2.get("sections", {}).items():
            print(f"   [{code}] cached={sec['cached']} (should be True)")


if __name__ == "__main__":
    asyncio.run(run())