#!/usr/bin/env python3 """Live integration smoke tests for Vision -> Mapping -> Grounding pipeline. Run from the backend package root:: cd backend python scripts/test_pipeline_live.py Requires OPENAI_API_KEY in the environment (or .env at repo root). """ from __future__ import annotations import io import logging import os import sys import tempfile import traceback from pathlib import Path from typing import Any from unittest.mock import patch # ── Path bootstrap (backend/scripts -> repo root) ───────────────────────────── REPO_ROOT = Path(__file__).resolve().parents[2] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) # Prefer live key from the shell or repo .env before importing cached settings. def _load_dotenv_file(path: Path) -> None: if not path.is_file(): return for line in path.read_text(encoding="utf-8").splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in stripped: continue key, _, value = stripped.partition("=") key = key.strip() value = value.strip().strip('"').strip("'") if key and key not in os.environ: os.environ[key] = value _load_dotenv_file(REPO_ROOT / ".env") _load_dotenv_file(REPO_ROOT / "backend" / ".env") _env_key = (os.environ.get("OPENAI_API_KEY") or "").strip() if _env_key: os.environ["OPENAI_API_KEY"] = _env_key from backend.config import get_settings, settings # noqa: E402 if _env_key: get_settings.cache_clear() settings.openai_api_key = _env_key from docx import Document # noqa: E402 from PIL import Image # noqa: E402 from backend.domain import template_discoverer # noqa: E402 from backend.ingest import pipeline as ingest # noqa: E402 from backend.llm import openai_client # noqa: E402 from backend.llm.openai_client import _parse_llm_json # noqa: E402 from backend.pipeline import section_mapper # noqa: E402 from backend.storage import photo_store # noqa: E402 # ── Console helpers ───────────────────────────────────────────────────────── GREEN = "\033[92m" RED = "\033[91m" YELLOW = "\033[93m" CYAN = "\033[96m" RESET = "\033[0m" _log_handler: logging.Handler | None = None _captured_logs: list[str] = [] def _pass(msg: str) -> None: print(f"{GREEN}[PASS]{RESET} {msg}") def _fail(msg: str) -> None: print(f"{RED}[FAIL]{RESET} {msg}") def _info(msg: str) -> None: print(f"{CYAN}[INFO]{RESET} {msg}") def _warn(msg: str) -> None: print(f"{YELLOW}[WARN]{RESET} {msg}") def _section(title: str) -> None: print(f"\n{'=' * 72}\n{title}\n{'=' * 72}") def _install_log_capture() -> None: global _log_handler, _captured_logs _captured_logs = [] class _ListHandler(logging.Handler): def emit(self, record: logging.LogRecord) -> None: _captured_logs.append(record.getMessage()) _log_handler = _ListHandler() _log_handler.setLevel(logging.INFO) logging.getLogger("backend.pipeline.section_mapper").addHandler(_log_handler) def _remove_log_capture() -> None: global _log_handler if _log_handler is not None: logging.getLogger("backend.pipeline.section_mapper").removeHandler(_log_handler) _log_handler = None def _log_contains(fragment: str) -> bool: return any(fragment in line for line in _captured_logs) def _tiny_jpeg() -> bytes: buf = io.BytesIO() Image.new("RGB", (64, 64), color=(180, 180, 180)).save(buf, format="JPEG") return buf.getvalue() def _write_fixture_docx(tmp: Path) -> tuple[Path, Path]: """Template + reference bundle with a known D2 baseline paragraph.""" tpl = tmp / "template.docx" doc = Document() doc.add_paragraph("D2 Roof coverings") doc.save(str(tpl)) past = tmp / "past.docx" doc2 = Document() doc2.add_paragraph("Section D2 - Roof Coverings") doc2.add_paragraph( "The roof covering comprised slate tiles laid to pitched timber rafters. " "Inspection was limited to ground level using binoculars." ) doc2.save(str(past)) return tpl, past def _bootstrap_tenant(tmp: Path, tenant: str) -> str: """Ingest template + reference; return draft id with one selected D2 photo.""" tpl, past = _write_fixture_docx(tmp) ingest.ingest_report_template(tenant, tpl) ingest.ingest_reference(tenant, past) template_discoverer.ensure_canonical_schema(tenant) draft = f"live-draft-{tenant}" row = photo_store.add_section_photo( tenant, draft, "D2", _tiny_jpeg(), content_type="image/jpeg", ) photo_store.set_ai_selection(tenant, draft, "D2", [row.id]) return draft def _d2_section(result: Any) -> Any: for section in result.sections: if section.section_id.upper() == "D2": return section raise AssertionError("D2 section missing from report result") def _guard_live_llm() -> bool: key = (os.environ.get("OPENAI_API_KEY") or settings.openai_api_key or "").strip() if not key: _fail("OPENAI_API_KEY is not set — export it or add it to .env at repo root.") return False settings.openai_api_key = key if not openai_client.is_available(): _fail("openai_client.is_available() is False after loading OPENAI_API_KEY.") return False _info(f"OpenAI client ready (key prefix: {key[:7]}…).") return True def test_case_a_pristine_passthrough(tmp: Path) -> bool: """Empty notes + zero photo observations → baseline passthrough, no auditor LLM.""" _section("TEST CASE A — THE PRISTINE PASSTHROUGH") tenant = "live-passthrough" draft = _bootstrap_tenant(tmp, tenant) baseline_snippet = "binoculars" stabilize_calls = 0 chat_text_calls = 0 chat_json_async_calls = 0 real_stabilize = section_mapper.stabilize_section real_chat_text = openai_client.chat_text real_chat_json_async = openai_client.chat_json_async async def _counting_stabilize(*args: Any, **kwargs: Any) -> Any: nonlocal stabilize_calls stabilize_calls += 1 return await real_stabilize(*args, **kwargs) def _counting_chat_text(*args: Any, **kwargs: Any) -> str: nonlocal chat_text_calls chat_text_calls += 1 return real_chat_text(*args, **kwargs) async def _counting_chat_json_async(*args: Any, **kwargs: Any) -> dict: nonlocal chat_json_async_calls chat_json_async_calls += 1 return await real_chat_json_async(*args, **kwargs) try: _install_log_capture() with patch( "backend.pipeline.section_mapper.vision_observations_for_section", return_value=([], None), ): with patch.object( section_mapper, "stabilize_section", side_effect=_counting_stabilize ): with patch.object( openai_client, "chat_text", side_effect=_counting_chat_text ): with patch.object( openai_client, "chat_json_async", side_effect=_counting_chat_json_async, ): result = section_mapper.generate_report( tenant, "", interference_level="minimum", report_draft_id=draft, only_section_ids=["D2"], ) section = _d2_section(result) checks: list[tuple[bool, str]] = [ (section.status == "OK", f"expected status OK, got {section.status!r}"), ( baseline_snippet in section.text.lower(), "section text should retain retrieved baseline prose", ), ( section.grounding_passed is True, f"expected grounding_passed=True, got {section.grounding_passed!r}", ), ( stabilize_calls == 0, f"stabilize_section must not run (calls={stabilize_calls})", ), ( chat_json_async_calls == 0, f"grounding auditor chat_json_async must not run (calls={chat_json_async_calls})", ), ( _log_contains("baseline_passthrough_no_observations"), "expected baseline_passthrough_no_observations log line", ), ] ok = True for passed, detail in checks: if passed: _pass(detail) else: _fail(detail) ok = False if chat_text_calls: _warn( f"chat_text invoked {chat_text_calls} time(s) " "(notes expansion / mapping should be skipped for empty observations)." ) else: _pass("no chat_text mapping calls during passthrough") return ok except Exception as exc: _fail(f"unexpected error: {exc}") traceback.print_exc() return False finally: _remove_log_capture() def test_case_b_obscured_limitations_pass(tmp: Path) -> bool: """Limitations-only vision payload → grounding accepts defensive limitation prose.""" _section("TEST CASE B — THE OBSCURED LIMITATIONS PASS") tenant = "live-limitations" draft = _bootstrap_tenant(tmp, tenant) limitation_phrase = "restricted view due to heavy insulation storage" vision_note = f"Photo limitations: {limitation_phrase}." try: with patch( "backend.pipeline.section_mapper.vision_observations_for_section", return_value=([], vision_note), ): result = section_mapper.generate_report( tenant, "", interference_level="maximum", report_draft_id=draft, only_section_ids=["D2"], ) section = _d2_section(result) text_lower = section.text.lower() checks: list[tuple[bool, str]] = [ (section.status == "OK", f"expected status OK, got {section.status!r}"), ( section.grounding_passed is True, f"expected grounding_passed=True, got {section.grounding_passed!r}", ), ( "insulation" in text_lower or "restricted" in text_lower or "view" in text_lower or limitation_phrase.lower() in text_lower, "section prose should reflect the photo limitation anchor", ), ( vision_note.split("Photo limitations:", 1)[1].strip().lower() in text_lower or any( token in text_lower for token in ("insulation", "restricted", "view") ), "grounding should not strip limitation-defensive language as hallucination", ), ] ok = True for passed, detail in checks: if passed: _pass(detail) else: _fail(detail) ok = False _info(f"D2 output excerpt: {section.text[:220]}…") return ok except Exception as exc: _fail(f"unexpected error: {exc}") traceback.print_exc() return False def test_case_c_json_fence_sanitization() -> bool: """_parse_llm_json strips markdown fences without JSONDecodeError.""" _section("TEST CASE C — JSON FENCE SANITIZATION TRAP") samples = [ '```json\n{"status": "ok"}\n```', '```\n{"status": "ok", "count": 1}\n```', '{"status": "bare"}', ] ok = True for index, raw in enumerate(samples, start=1): try: parsed = _parse_llm_json(raw) if not isinstance(parsed, dict): _fail(f"sample {index}: expected dict, got {type(parsed).__name__}") ok = False continue if parsed.get("status") != "ok" and index < 3: _fail(f"sample {index}: unexpected payload {parsed!r}") ok = False continue if index == 3 and parsed != {"status": "bare"}: _fail(f"sample {index}: unexpected payload {parsed!r}") ok = False continue _pass(f"sample {index}: fence-safe parse → {parsed!r}") except Exception as exc: _fail(f"sample {index}: raised {exc!r}") traceback.print_exc() ok = False # Optional live chat_json path: monkey-patch completion to return fenced JSON. if openai_client.is_available(): try: fenced_payload = ( '```json\n{"passed": true, "_audit_summary": "fence trap"}\n```' ) def _fake_create(**kwargs: Any) -> Any: class _Msg: content = fenced_payload class _Choice: message = _Msg() class _Resp: choices = [_Choice()] return _Resp() with patch.object( openai_client._get_client().chat.completions, "create", _fake_create ): live = openai_client.chat_json( [{"role": "user", "content": "return json"}], model=settings.grounding_model, ) if live.get("passed") is True: _pass("chat_json live wrapper parsed fenced auditor JSON") else: _fail(f"chat_json live wrapper returned unexpected payload: {live!r}") ok = False except Exception as exc: _fail(f"chat_json fenced integration check failed: {exc}") traceback.print_exc() ok = False else: _warn("Skipping live chat_json fence trap (OpenAI unavailable).") return ok def main() -> int: print(f"{CYAN}RICS pipeline live integration runner{RESET}") print(f"Repo root: {REPO_ROOT}") if not _guard_live_llm(): return 1 results: dict[str, bool] = {} with tempfile.TemporaryDirectory(prefix="rics-live-") as tmp_name: tmp = Path(tmp_name) settings.data_dir = str(tmp / "data") _info(f"Using isolated data_dir={settings.data_dir}") results["A_pristine_passthrough"] = test_case_a_pristine_passthrough(tmp) results["B_obscured_limitations"] = test_case_b_obscured_limitations_pass(tmp) results["C_json_fence_sanitization"] = test_case_c_json_fence_sanitization() _section("SUMMARY") failed = [name for name, ok in results.items() if not ok] for name, ok in results.items(): if ok: _pass(name) else: _fail(name) if failed: print(f"\n{RED}{len(failed)} test block(s) failed:{RESET} {', '.join(failed)}") return 1 print(f"\n{GREEN}All pipeline live checks passed.{RESET}") return 0 if __name__ == "__main__": sys.exit(main())