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" def test_ragcore_provider_degrades_to_unavailable(monkeypatch): def fake_client(*args, **kwargs): raise httpx.ConnectError("no ragcore in this environment") provider = RAGcoreKnowledgeProvider() monkeypatch.setattr(provider, "_client", fake_client) answer = provider.ask("Anything?", "test-correlation-3") assert answer.evidence_state == "unavailable" assert answer.sources == []