Spaces:
Runtime error
Runtime error
File size: 4,488 Bytes
7d37f11 3c31a2a 7d37f11 3c31a2a 7d37f11 3c31a2a 7d37f11 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """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())
|