MobilityOps/backend/tests/test_knowledge.py

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 == []