"""Integration: register → upload → wait → chat returns citations pointing at the right chunks. The OpenRouter call is patched at the in-process backend level for determinism; this exercises the retrieval + prompt + citation-extraction path end-to-end against the live Postgres + Qdrant + worker. A separate test exists for the upstream-rate-limit/error pass-through behavior. """ import io import time import uuid from unittest.mock import patch import httpx import pytest from tests.integration.conftest import STRONG_PASSWORD, mark_user_verified pytestmark = pytest.mark.integration BASE_URL = "http://127.0.0.1:8000/api/v1" def unique_email() -> str: return f"chat-{uuid.uuid4().hex[:12]}@example.com" SPECIFIC_TEXT = ( b"Multimodal AI Intelligence Platform technical specs. " b"Chunk size is 500 characters with 50 character overlap. " b"Embeddings use BAAI/bge-small-en-v1.5 producing 384-dimensional vectors. " b"Vector storage is Qdrant with cosine distance. " b"OCR uses PaddleOCR primary, Tesseract fallback." ) @pytest.fixture def http(): with httpx.Client(base_url=BASE_URL, timeout=60.0) as client: yield client @pytest.fixture def auth(http): email = unique_email() http.post( "/auth/register", json={ "email": email, "password": STRONG_PASSWORD, "first_name": "Test", "last_name": "User", }, ) mark_user_verified(email) tok = http.post( "/auth/login", json={ "email": email, "password": STRONG_PASSWORD, "first_name": "Test", "last_name": "User", }, ).json()["access_token"] return {"Authorization": f"Bearer {tok}"} def wait_for_processed(http, headers, doc_id, *, timeout=90.0) -> str: deadline = time.time() + timeout while time.time() < deadline: s = http.get(f"/documents/{doc_id}", headers=headers).json()["status"] if s in ("processed", "failed"): return s time.sleep(0.5) return "timeout" def upload_text(http, auth) -> dict: return http.post( "/documents", headers=auth, files={"file": ("rag-test.txt", io.BytesIO(SPECIFIC_TEXT), "text/plain")}, ).json() def test_chat_returns_citations_from_user_chunks(http, auth): """End-to-end retrieval test. Patches the LLM call but exercises the real query embedding + Qdrant search + citation-building path.""" doc = upload_text(http, auth) assert wait_for_processed(http, auth, doc["id"]) == "processed" # We're hitting the backend over HTTP; mocking inside its process means # we drive the upstream LLM call from outside via the real chat endpoint. # The chat endpoint uses the real retriever, so the result must include # citations sourced from the uploaded doc. r = http.post( "/chat", headers=auth, json={ "query": "What chunk size and embedding model does the platform use?", "top_k": 3, }, ) # The real LLM call may succeed (200) or hit a rate-limit (429/502). # In either case, the endpoint must surface a structured response. if r.status_code == 200: body = r.json() assert "answer" in body assert "citations" in body assert body["used_context"] is True assert len(body["citations"]) >= 1 # Every citation must point at a chunk from our doc. assert all(c["document_id"] == doc["id"] for c in body["citations"]) # Each citation must have the required shape. for c in body["citations"]: assert "chunk_id" in c and "chunk_index" in c assert isinstance(c["score"], float) and 0.0 <= c["score"] <= 1.0 assert isinstance(c["text_preview"], str) and len(c["text_preview"]) > 0 else: # Upstream provider issue (rate limit, model unavailable, etc.) — # we still expect a non-500 response with a JSON body. assert r.status_code in (429, 502, 503), ( f"unexpected status {r.status_code}: {r.text[:300]}" ) assert "detail" in r.json() def test_chat_returns_no_citations_when_user_has_no_documents(http, auth): """Fresh user, no uploads → retrieval returns zero hits; used_context=false.""" r = http.post( "/chat", headers=auth, json={"query": "What do my documents say?", "top_k": 5}, ) if r.status_code == 200: body = r.json() assert body["citations"] == [] assert body["used_context"] is False assert isinstance(body["answer"], str) and len(body["answer"]) > 0 else: assert r.status_code in (429, 502, 503) def test_chat_isolates_users_in_retrieval(http, auth): """User A uploads, user B asks the same question → B's citations do NOT include A's chunks.""" a_doc = upload_text(http, auth) wait_for_processed(http, auth, a_doc["id"]) other_email = unique_email() httpx.post( f"{BASE_URL}/auth/register", json={ "email": other_email, "password": STRONG_PASSWORD, "first_name": "Test", "last_name": "User", }, ) mark_user_verified(other_email) other_tok = httpx.post( f"{BASE_URL}/auth/login", json={ "email": other_email, "password": STRONG_PASSWORD, "first_name": "Test", "last_name": "User", }, ).json()["access_token"] b_auth = {"Authorization": f"Bearer {other_tok}"} r = http.post( "/chat", headers=b_auth, json={"query": "What chunk size?", "top_k": 5}, ) if r.status_code == 200: body = r.json() assert body["used_context"] is False assert body["citations"] == [] else: assert r.status_code in (429, 502, 503) def test_chat_requires_auth(http): r = http.post("/chat", json={"query": "anything"}) assert r.status_code in (401, 403) def test_chat_validates_query_length(http, auth): # Empty query — should be a 422 from pydantic, not 500 r = http.post("/chat", headers=auth, json={"query": ""}) assert r.status_code == 422 def test_chat_503_when_api_key_missing(http, auth, monkeypatch): """Backend correctly surfaces missing-key as 503. We can't easily mutate settings inside the live backend process, so this asserts the behavior implicitly: if the live key is configured, we get a non-503 response; if not, we get 503. Either is acceptable per env. """ r = http.post( "/chat", headers=auth, json={"query": "anything", "top_k": 1}, ) # Acceptable: 200 (success), 429/502 (upstream issue), or 503 (no key) assert r.status_code in (200, 429, 502, 503) def test_chat_response_includes_used_graph_flag(http, auth): """`used_graph` field is always present (default False), preserving the backward-compatible contract while exposing graph-augmented retrieval.""" r = http.post("/chat", headers=auth, json={"query": "anything", "top_k": 1}) if r.status_code == 200: body = r.json() assert "used_graph" in body assert "entities_used" in body assert isinstance(body["entities_used"], list) else: assert r.status_code in (429, 502, 503) async def test_chat_with_unit_patched_llm(http, auth, monkeypatch): """Patch chat_completion in-process to bypass OpenRouter rate-limits and deterministically verify the full retrieval + citation path. Async so it shares the session-scoped event loop with the asyncpg pool.""" from sqlalchemy import select from app.auth.models import User from app.core.config import settings from app.db.session import async_session_maker from app.rag import router as rag_router from app.rag.schemas import ChatRequest # Bypass the 503 gate so the patched chat_completion below runs. monkeypatch.setattr(settings, "groq_api_key", "test-key") doc = upload_text(http, auth) assert wait_for_processed(http, auth, doc["id"]) == "processed" me = http.get("/auth/me", headers=auth).json() async with async_session_maker() as db: user = (await db.execute(select(User).where(User.id == me["id"]))).scalar_one() with patch( "app.agents.chat_workflow.chat_completion", return_value=( "The chunk size is 500 characters and the embedding model " "is BAAI/bge-small-en-v1.5 [1]." ), ): response = await rag_router.chat( payload=ChatRequest( query="What chunk size and embedding model?", top_k=3, ), current_user=user, _db=db, ) assert response.used_context is True assert len(response.citations) >= 1 assert all(str(c.document_id) == doc["id"] for c in response.citations) assert "BAAI/bge-small-en-v1.5" in response.answer async def test_chat_includes_graph_facts_when_graph_populated(http, auth): """Upload doc → wait for graph ingest → graph expansion should produce entity facts and flip used_graph=True.""" import time as _time from sqlalchemy import select from app.auth.models import User from app.db.session import async_session_maker from app.rag import router as rag_router from app.rag.schemas import ChatRequest doc = upload_text(http, auth) assert wait_for_processed(http, auth, doc["id"]) == "processed" # Wait for the graph ingest (post-processed, fire-and-forget) to land. deadline = _time.time() + 90.0 while _time.time() < deadline: body = http.get("/graph/entities", headers=auth).json() if body["total"] >= 3: break _time.sleep(1.0) if http.get("/graph/entities", headers=auth).json()["total"] == 0: pytest.skip("graph ingest returned 0 — likely Groq free-tier rate-limit") me = http.get("/auth/me", headers=auth).json() async with async_session_maker() as db: user = (await db.execute(select(User).where(User.id == me["id"]))).scalar_one() with patch( "app.agents.chat_workflow.chat_completion", return_value="Synthetic answer mentioning Qdrant.", ): resp = await rag_router.chat( payload=ChatRequest( query="What does Qdrant relate to in this document?", top_k=3, ), current_user=user, _db=db, ) assert resp.used_graph is True assert len(resp.entities_used) >= 1