File size: 6,069 Bytes
7be0127 80b7807 7be0127 80b7807 4a474b7 0b698f0 7be0127 7fdbe89 7be0127 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | """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"]))
|