from __future__ import annotations import re from pathlib import Path import httpx from app.core.config import get_settings from app.services.knowledge.demo import DemoKnowledgeProvider from app.services.knowledge.ragcore import RAGcoreKnowledgeProvider def test_brief_exact_damage_question_in_all_three_languages(): # The exact validation questions from docs/fleet-ops-correction/current-gap-audit.md # -- each must ground on the damage procedure as its *primary* (top-ranked) source, # not merely appear somewhere in the top-3, and the source/version/section/excerpt # must all come from that same-language document (never an English fallback). provider = DemoKnowledgeProvider() cases = { "nl-BE": "Wat moet ik doen wanneer een voertuig beschadigd terugkomt?", "en-GB": "What should I do when a vehicle returns with damage?", "fr-BE": "Que dois-je faire lorsqu'un véhicule revient endommagé ?", } for language, question in cases.items(): answer = provider.ask(question, f"test-brief-{language}", language) assert answer.evidence_state == "grounded", language assert answer.sources, language assert answer.sources[0].document_id == "damage-procedure", ( f"{language}: expected the damage procedure as the primary source, " f"got {answer.sources[0].document_id!r}" ) assert answer.answer assert answer.sources[0].excerpt def test_knowledge_procedures_never_mention_mobilityops_or_poc(): # Section 2 of docs/fleet-ops-correction/current-gap-audit.md: the visible brand # name is exactly "Fleet Ops", and "PoC" must never appear in visible content -- # including the demo knowledge base, not just the frontend. procedures_dir = Path(get_settings().knowledge_dir) offenders = [] for path in sorted(procedures_dir.glob("*/*.md")): text = path.read_text(encoding="utf-8") if "MobilityOps" in text or re.search(r"\bPoC\b", text): offenders.append(str(path)) assert offenders == [] def test_s6_damage_question_is_grounded_with_expected_sources(): provider = DemoKnowledgeProvider() answer = provider.ask( "What must I do when a vehicle returns with damage?", "test-correlation-1" ) assert answer.evidence_state == "grounded" document_ids = {s.document_id for s in answer.sources} assert "damage-procedure" in document_ids assert "vehicle-return-procedure" in document_ids assert answer.answer # never empty for a grounded answer for source in answer.sources: assert source.excerpt def test_unrelated_question_is_insufficient(): provider = DemoKnowledgeProvider() answer = provider.ask("What is the capital of France?", "test-correlation-2") assert answer.evidence_state == "insufficient" assert "France" not in answer.answer # never fabricates an answer beyond the procedures def test_demo_provider_health_reports_document_count(): provider = DemoKnowledgeProvider() health = provider.health() assert health.provider == "demo" assert health.available is True assert health.document_count == 11 def test_demo_provider_health_reports_document_count_per_language(): provider = DemoKnowledgeProvider() for language in ("nl-BE", "en-GB", "fr-BE"): assert provider.health(language).document_count == 11 def test_demo_provider_grounds_damage_question_in_dutch(): provider = DemoKnowledgeProvider() answer = provider.ask( "Wat moet ik doen als een voertuig terugkomt met schade?", "test-correlation-nl", "nl-BE", ) assert answer.evidence_state == "grounded" document_ids = {s.document_id for s in answer.sources} assert "damage-procedure" in document_ids def test_demo_provider_grounds_damage_question_in_french(): provider = DemoKnowledgeProvider() answer = provider.ask( "Que dois-je faire quand un véhicule revient avec des dommages ?", "test-correlation-fr", "fr-BE", ) assert answer.evidence_state == "grounded" document_ids = {s.document_id for s in answer.sources} assert "damage-procedure" in document_ids def test_demo_provider_insufficient_evidence_message_is_localized(): provider = DemoKnowledgeProvider() nl_answer = provider.ask( "Wat is de hoofdstad van Frankrijk?", "test-correlation-nl-2", "nl-BE" ) fr_answer = provider.ask( "Quelle est la capitale de la France ?", "test-correlation-fr-2", "fr-BE" ) assert nl_answer.evidence_state == "insufficient" assert fr_answer.evidence_state == "insufficient" assert nl_answer.answer != fr_answer.answer assert "France" not in nl_answer.answer assert "France" not in fr_answer.answer def test_ask_question_endpoint_grounded(ops_client): response = ops_client.post( "/api/v1/knowledge/questions", json={"question": "What must I do when a vehicle returns with damage?"}, ) assert response.status_code == 200 body = response.json() assert body["evidence_state"] == "grounded" assert body["provider"] == "demo" assert len(body["sources"]) > 0 def test_ask_question_requires_authentication(client): response = client.post("/api/v1/knowledge/questions", json={"question": "Anything?"}) assert response.status_code == 401 def test_ask_question_is_audited_without_leaking_full_text(ops_client): ops_client.post( "/api/v1/knowledge/questions", json={"question": "What must I do when a vehicle returns with damage?"}, ) events = ops_client.get( "/api/v1/audit", params={"action": "knowledge_question_asked"} ).json() assert len(events) >= 1 metadata = events[0]["metadata"] assert "evidence_state" in metadata assert "source_ids" in metadata assert "question" not in metadata def test_knowledge_status_endpoint(ops_client): response = ops_client.get("/api/v1/knowledge/status") assert response.status_code == 200 assert response.json()["provider"] == "demo" class _FakeResponse: def __init__(self, status_code: int, body: dict): self.status_code = status_code self._body = body def json(self) -> dict: return self._body class _FakeClient: def __init__(self, get_response=None, post_response=None, post_responses=None, raise_on=None): self._get_response = get_response self._post_response = post_response # Maps a path (e.g. "/v1/search") to its own response, for tests that need # /v1/answers and /v1/search to behave differently in the same call. Falls back # to the single post_response when a path has no specific entry, so every # existing single-endpoint test keeps working unchanged. self._post_responses = post_responses or {} self._raise_on = raise_on def __enter__(self): return self def __exit__(self, *args): return False def get(self, path): if self._raise_on == "get": raise httpx.ConnectError("no ragcore in this environment") return self._get_response def post(self, path, json=None): if self._raise_on == "post": raise httpx.ConnectError("no ragcore in this environment") return self._post_responses.get(path, self._post_response) def test_ragcore_provider_degrades_to_unavailable(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1") monkeypatch.setattr(provider, "_client", lambda: _FakeClient(raise_on="post")) answer = provider.ask("Anything?", "test-correlation-3") assert answer.evidence_state == "unavailable" assert answer.sources == [] def test_ragcore_provider_without_configured_space_is_unavailable_without_a_network_call( monkeypatch, ): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider._settings, "ragcore_space_id", "") def fail_if_called(): raise AssertionError("should not call RAGcore without a configured space id") monkeypatch.setattr(provider, "_client", fail_if_called) answer = provider.ask("Anything?", "test-correlation-no-space") assert answer.evidence_state == "unavailable" def test_ragcore_provider_health_reports_ready_status(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr( provider, "_client", lambda: _FakeClient(get_response=_FakeResponse(200, {"status": "ok"})), ) health = provider.health() assert health.provider == "ragcore" assert health.available is True def test_ragcore_provider_health_reports_degraded_status(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr( provider, "_client", lambda: _FakeClient(get_response=_FakeResponse(200, {"status": "degraded"})), ) health = provider.health() assert health.available is False def test_ragcore_provider_health_degrades_on_connection_error(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider, "_client", lambda: _FakeClient(raise_on="get")) health = provider.health() assert health.available is False assert "unavailable" in health.detail.lower() def _answers_body(**overrides) -> dict: body = { "answer": "Report damage and route the vehicle to maintenance.", "answerability": "answerable", "citations": [ { "id": "cite-1", "document_id": "doc-1", "document_version_id": "version-1", "title": "Damage handling procedure", "section": "Detection", "excerpt": "Inspect the vehicle for visible damage.", } ], } body.update(overrides) return body def test_ragcore_provider_grounded_answer_maps_citations_to_sources(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1") monkeypatch.setattr( provider, "_client", lambda: _FakeClient(post_response=_FakeResponse(200, _answers_body())), ) answer = provider.ask("What must I do about damage?", "test-correlation-grounded") assert answer.evidence_state == "grounded" assert answer.answer assert len(answer.sources) == 1 source = answer.sources[0] assert source.document_id == "doc-1" assert source.title == "Damage handling procedure" assert source.version == "version-1" assert source.section == "Detection" assert source.excerpt def test_ragcore_provider_not_answerable_is_insufficient_and_never_fabricates(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1") monkeypatch.setattr( provider, "_client", lambda: _FakeClient( post_response=_FakeResponse( 200, _answers_body( answer="This should never be shown.", answerability="not_answerable", citations=[], ), ) ), ) answer = provider.ask("Unrelated question?", "test-correlation-insufficient") assert answer.evidence_state == "insufficient" assert answer.answer == "" assert answer.sources == [] def test_ragcore_provider_answerable_without_citations_is_insufficient(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1") monkeypatch.setattr( provider, "_client", lambda: _FakeClient( post_response=_FakeResponse( 200, _answers_body(answerability="answerable", citations=[]) ) ), ) answer = provider.ask("What must I do about damage?", "test-correlation-no-citations") assert answer.evidence_state == "insufficient" assert answer.sources == [] def test_ragcore_provider_non_200_response_is_unavailable(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1") monkeypatch.setattr( provider, "_client", lambda: _FakeClient(post_response=_FakeResponse(401, {"code": "AUTHENTICATION_REQUIRED"})), ) answer = provider.ask("Anything?", "test-correlation-401") assert answer.evidence_state == "unavailable" def test_ragcore_provider_malformed_response_is_unavailable(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1") # A malformed /v1/answers body triggers the same search fallback a real outage # would, so the fallback's own /v1/search response must also be malformed here to # exercise "the whole backend is misbehaving, not just one endpoint" honestly. monkeypatch.setattr( provider, "_client", lambda: _FakeClient( post_responses={ "/v1/answers": _FakeResponse(200, {"citations": "not-a-list"}), "/v1/search": _FakeResponse(200, {"results": "not-a-list"}), } ), ) answer = provider.ask("Anything?", "test-correlation-malformed") assert answer.evidence_state == "unavailable" def _search_body(**overrides) -> dict: body = { "results": [ { "chunk_id": "chunk-1", "citation": { "id": "cite-1", "document_id": "doc-1", "document_version_id": "version-1", "title": "Vehicle return procedure", "section": "Return", "excerpt": "Register the return odometer reading before releasing the vehicle.", }, "rank": 1, "scores": {"dense": None, "sparse": None, "fused": 0.5, "rerank": None}, } ], "degraded": False, } body.update(overrides) return body def test_ragcore_provider_falls_back_to_search_when_answers_unavailable(monkeypatch): """/v1/answers itself failing (a real RAGcore-side outage in its generation step, not a real 'insufficient evidence' classification) must not silently degrade straight to 'unavailable' when RAGcore's own retrieval still works -- it should show the real, cited excerpt search actually found instead.""" provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1") monkeypatch.setattr( provider, "_client", lambda: _FakeClient( post_responses={ "/v1/answers": _FakeResponse(503, {"code": "VALIDATION_RETRIES_EXHAUSTED"}), "/v1/search": _FakeResponse(200, _search_body()), } ), ) answer = provider.ask("What is the vehicle return procedure?", "test-correlation-fallback") assert answer.evidence_state == "grounded" assert "Register the return odometer reading" in answer.answer assert "Vehicle return procedure" in answer.answer assert len(answer.sources) == 1 assert answer.sources[0].title == "Vehicle return procedure" assert answer.sources[0].excerpt == ( "Register the return odometer reading before releasing the vehicle." ) def test_ragcore_provider_fallback_answer_is_localized(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1") monkeypatch.setattr( provider, "_client", lambda: _FakeClient( post_responses={ "/v1/answers": _FakeResponse(503, {"code": "VALIDATION_RETRIES_EXHAUSTED"}), "/v1/search": _FakeResponse(200, _search_body()), } ), ) answer = provider.ask( "Wat is de procedure voor een voertuigretour?", "test-correlation-fallback-nl", language="nl-BE", ) assert answer.evidence_state == "grounded" assert answer.answer.startswith("Volgens ") def test_ragcore_provider_fallback_with_no_search_results_is_insufficient(monkeypatch): provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1") monkeypatch.setattr( provider, "_client", lambda: _FakeClient( post_responses={ "/v1/answers": _FakeResponse(503, {"code": "VALIDATION_RETRIES_EXHAUSTED"}), "/v1/search": _FakeResponse(200, _search_body(results=[])), } ), ) answer = provider.ask("Unrelated question?", "test-correlation-fallback-empty") assert answer.evidence_state == "insufficient" assert answer.answer == "" assert answer.sources == []