File size: 6,072 Bytes
8b96826 | 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 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | """The app must degrade gracefully, never crash: input validation,
fail-open cache, broad LLM fallback, global exception handler."""
import json
import pytest
import requests as requests_lib
from fastapi.testclient import TestClient
import main
from app import cache
from app.agents import synthesizer
from app.providers import gemini, groq, llm
from app.providers.gemini import ProviderError
from tests.test_synthesizer import make_context
client = TestClient(main.app)
crash_client = TestClient(main.app, raise_server_exceptions=False)
def stream_payloads(response):
return [
json.loads(line[6:])
for line in response.text.splitlines()
if line.startswith("data: ")
]
# ---- plan stream input validation ------------------------------------------
def test_invalid_arrive_by_returns_friendly_sse_error():
response = client.get(
"/api/plan/stream",
params={"from": "Aliganj", "to": "Chowk", "arrive_by": "not-a-date"},
)
assert response.status_code == 200 # SSE, not a 500
payloads = stream_payloads(response)
assert payloads[0]["type"] == "error"
assert "invalid" in payloads[0]["message"].lower()
assert payloads[-1]["type"] == "done"
def test_past_arrive_by_returns_friendly_sse_error():
response = client.get(
"/api/plan/stream",
params={"from": "Aliganj", "to": "Chowk", "arrive_by": "2020-01-01T09:30"},
)
payloads = stream_payloads(response)
assert payloads[0]["type"] == "error"
assert "past" in payloads[0]["message"].lower()
def test_far_future_arrive_by_returns_friendly_sse_error():
response = client.get(
"/api/plan/stream",
params={"from": "Aliganj", "to": "Chowk", "arrive_by": "2099-01-01T09:30"},
)
payloads = stream_payloads(response)
assert payloads[0]["type"] == "error"
assert "week" in payloads[0]["message"].lower()
# ---- endpoint crash safety ----------------------------------------------------
def test_autocomplete_crash_returns_empty_not_500(monkeypatch):
def boom(q):
raise RuntimeError("unexpected bug key=secret123")
monkeypatch.setattr(main.geocode, "autocomplete_place", boom)
response = client.get("/api/autocomplete", params={"q": "hazra"})
assert response.status_code == 200
assert response.json() == {"suggestions": []}
def test_ask_crash_returns_clean_500_without_secrets(monkeypatch):
monkeypatch.setattr(main.llm, "available", lambda: True)
def boom(question):
raise RuntimeError("boom at url?key=topsecret99")
monkeypatch.setattr(main.orchestrator, "agent_ask", boom)
response = client.post("/api/ask", json={"question": "hello"})
assert response.status_code == 500
assert "topsecret99" not in response.text
assert "detail" in response.json()
def test_ask_empty_question_rejected(monkeypatch):
monkeypatch.setattr(main.llm, "available", lambda: True)
response = client.post("/api/ask", json={"question": " "})
assert response.status_code == 422
def test_global_handler_catches_unhandled_route_errors(monkeypatch):
def boom():
raise ValueError("internal bug key=hidden42")
monkeypatch.setattr(main.llm, "available", boom)
response = crash_client.get("/api/health")
assert response.status_code == 500
assert "hidden42" not in response.text
assert "detail" in response.json()
# ---- fail-open cache ----------------------------------------------------------
def test_cache_read_failure_is_a_miss(monkeypatch):
def broken_conn():
raise RuntimeError("disk locked")
monkeypatch.setattr(cache, "_get_conn", broken_conn)
assert cache.get("anything") is None # no exception
def test_cache_write_failure_is_silent(monkeypatch):
def broken_conn():
raise RuntimeError("disk full")
monkeypatch.setattr(cache, "_get_conn", broken_conn)
cache.set("key", "value", ttl_seconds=60) # must not raise
def test_cached_function_still_works_when_cache_broken(monkeypatch):
def broken_conn():
raise RuntimeError("corrupted db")
monkeypatch.setattr(cache, "_get_conn", broken_conn)
@cache.cached(ttl_seconds=60)
def compute(x):
return x * 2
assert compute(21) == 42 # function result delivered despite broken cache
# ---- LLM layer degradation ----------------------------------------------------
def test_synthesizer_survives_unexpected_llm_crash(monkeypatch, fake_keys):
def boom(*args, **kwargs):
raise ValueError("adapter bug, not a ProviderError")
monkeypatch.setattr(synthesizer.llm, "chat", boom)
verdict, provider = synthesizer.synthesize(make_context())
assert provider == "deterministic"
assert verdict["recommended_departure"] == "08:15"
def test_llm_falls_back_when_gemini_adapter_crashes(monkeypatch, fake_keys):
def gemini_bug(*args, **kwargs):
raise KeyError("unexpected response shape")
monkeypatch.setattr(gemini, "chat", gemini_bug)
monkeypatch.setattr(
groq, "chat", lambda *a, **k: {"text": "ok", "tool_calls": [], "provider": "groq"}
)
result = llm.chat([{"role": "user", "content": "hi"}])
assert result["provider"] == "groq"
def test_gemini_non_json_response_is_provider_error(monkeypatch, fake_keys):
class BadResponse:
status_code = 200
text = "<html>gateway error</html>"
def json(self):
raise ValueError("not json")
monkeypatch.setattr(gemini.requests, "post", lambda *a, **k: BadResponse())
with pytest.raises(ProviderError, match="non-JSON"):
gemini.chat([{"role": "user", "content": "hi"}])
def test_groq_non_json_response_is_provider_error(monkeypatch, fake_keys):
class BadResponse:
status_code = 200
text = "<html>cloudflare</html>"
def json(self):
raise ValueError("not json")
monkeypatch.setattr(groq.requests, "post", lambda *a, **k: BadResponse())
with pytest.raises(ProviderError, match="non-JSON"):
groq.chat([{"role": "user", "content": "hi"}])
|