tests/test_api.py

"""HTTP aliases and readiness — uses the live orchestrator, no fake retrieval."""

import pytest
from fastapi.testclient import TestClient

import svarasetu.settings as config
from svarasetu.serve import app


@pytest.fixture(scope="module")
def client():
    with TestClient(app) as c:
        yield c


def test_health_and_ready(client):
    h = client.get("/health")
    assert h.status_code == 200
    body = h.json()
    assert "en" in body["configured_languages"]
    assert body["indexes_loaded"]
    r = client.get("/ready")
    assert r.status_code in (200, 503)
    if r.status_code == 200:
        assert r.json()["ready"] is True


def test_languages(client):
    data = client.get("/languages").json()
    assert data["active_languages"] == list(config.REGISTRY_ORDER)
    assert data["indexed_languages"] == list(config.REGISTRY_ORDER)


def test_api_rag_query_text(client):
    res = client.post(
        "/api/rag/query",
        json={"text": "What are the four chambers of the human heart?", "language_hint": "en", "bypass_gemini": True, "bypass_cache": True},
    )
    assert res.status_code == 200
    data = res.json()
    assert data["request_id"]
    assert data["language_detected"] == "en"
    assert "rag_core_ms" in data
    assert data["answer"]
    assert isinstance(data.get("stage_timings"), list)


def test_api_hindi_heart_question_returns_grounded_answer(client):
    res = client.post(
        "/api/rag/query",
        json={
            "text": "हृदय के चार कक्ष कौन से होते हैं?",
            "language_hint": "hi",
            "bypass_gemini": True,
            "bypass_cache": True,
        },
    )
    assert res.status_code == 200
    data = res.json()
    assert data["answer_source"] != "declined"
    assert data["language_detected"] == "hi"
    assert all(term in data["answer"] for term in ("दायाँ आलिंद", "दायाँ निलय", "बायाँ आलिंद", "बायाँ निलय"))
    assert data["guardrail_flags"]["grounding_passed"] is True


def test_query_alias_still_works(client):
    res = client.post("/query", json={"text": "What is a corporation?", "bypass_gemini": True, "bypass_cache": True})
    assert res.status_code == 200
    data = res.json()
    assert data["answer_source"] != "declined"
    assert "legal entity" in data["answer"].lower()
    assert "separate from its owners" in data["answer"].lower()


def test_query_schema_accepts_per_request_gemini_setting(client):
    # bypass_gemini keeps this API-contract test deterministic and network-free.
    res = client.post(
        "/api/rag/query",
        json={
            "text": "What is a corporation?",
            "gemini_enabled": True,
            "bypass_gemini": True,
        },
    )
    assert res.status_code == 200


def test_api_voice_query_without_audio_is_400(client):
    res = client.post("/api/voice/query", data={})
    assert res.status_code == 400


def test_ingest_alias_exists(client):
    res = client.post("/ingest", json={"text": "x"})
    assert res.status_code == 400


def test_home_ui_and_styles(client):
    home = client.get("/")
    assert home.status_code == 200
    assert "SVARASETU" in home.text
    assert "/static/styles.css" in home.text
    css = client.get("/static/styles.css")
    assert css.status_code == 200
    assert "--yellow" in css.text or "--gold" in css.text
    assert "Imbue" in css.text or "--font-d" in css.text
    assert "result-rail" in css.text
    assert "ingestBtn" not in home.text
    assert "Evidence-backed answers across 15 indexed languages" not in home.text
    assert 'id="geminiToggle"' in home.text
    assert "Gemini + RAG" in home.text
    assert "Search your RAG first" in home.text
    assert "lang-btn" not in home.text
    assert "Tap once. Then talk naturally." in home.text
    assert "SILENCE_STOP_MS" in home.text
    assert "vadHeardSpeech" in home.text
    assert "scheduleVoiceResume" in home.text
    assert "voiceSessionToggle" in home.text
    assert "The microphone pauses while SvaraSetu speaks." in home.text


def test_health_exposes_minilm_without_swapping_e5(client):
    body = client.get("/health").json()
    assert "e5" in (body.get("embedding_model") or "").lower()
    assert body.get("embedding_prefixes") is True
    assert "MiniLM" in (body.get("minilm_model") or "")
    assert body.get("rag_answer_mode") == "local_extractive"
    assert body.get("rag_answer_network_free") is True
    assert body.get("benchmark_contract") == "p100-under-50-v2"


def test_api_exposes_measured_hybrid_retrieval_breakdown(client):
    res = client.post(
        "/api/rag/query",
        json={
            "text": "What are the four chambers of the human heart?",
            "language_hint": "en",
            "bypass_gemini": True,
            "bypass_cache": True,
        },
    )
    assert res.status_code == 200
    timings = {row["stage"]: row for row in res.json()["stage_timings"]}
    for stage in ("dense_search", "lexical_search", "rrf_fusion"):
        assert stage in timings
        assert timings[stage]["ms"] >= 0
        assert timings[stage]["success"] is True


def test_live_benchmark_uses_strict_p100_goal(client):
    data = client.get("/benchmark?n=8").json()
    assert data["contract"] == "p100-under-50-v2"
    assert data["target_ms"] == 50.0
    assert data["pass"] is (data["p100"] <= data["target_ms"])
    assert data["sla_pass"] is (data["p100"] <= data["task_sla_ms"])
    assert len(data["samples"]) == 8
    assert round(data["worst_query"]["ms"], 2) == data["p100"]


def test_verified_cache_hit_keeps_evidence_references(client):
    payload = {
        "text": "What was the purpose of the Manhattan Project?",
        "language_hint": "en",
    }
    client.post("/api/rag/query", json=payload)
    data = client.post("/api/rag/query", json=payload).json()
    assert "cache" in data["answer_source"]
    assert data["retrieved_chunks"]
    assert len(data["citations"]) == min(3, len(data["retrieved_chunks"]))