Spaces:
Sleeping
Sleeping
File size: 2,663 Bytes
5aaf5ba | 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 | from collections.abc import AsyncIterator
from gargi_ai.main import app
from gargi_ai.providers import LLMProvider, TTSProvider
from gargi_ai.schemas import TeachingPayload
class FailingLLM(LLMProvider):
async def stream_explanation(self, prompt: str) -> AsyncIterator[str]:
raise RuntimeError("provider unavailable")
yield
async def generate_artifact(self, prompt: str) -> TeachingPayload:
raise RuntimeError("not reached")
class FailingTTS(TTSProvider):
async def synthesize(self, text: str, voice: str):
raise RuntimeError("tts unavailable")
class FailingArtifactLLM(LLMProvider):
async def stream_explanation(self, prompt: str) -> AsyncIterator[str]:
yield "The teacher response still works."
async def generate_artifact(self, prompt: str) -> TeachingPayload:
raise RuntimeError("artifact model overloaded")
def create_lesson(client):
return client.post(
"/api/v1/sessions", json={"topic": "Physics"}
).json()["id"]
def test_gemini_failure_is_a_recoverable_sse_error(client):
session_id = create_lesson(client)
app.state.llm = FailingLLM()
response = client.post(
f"/api/v1/sessions/{session_id}/teach",
json={"text": "Teach me."},
)
assert response.status_code == 200
assert "PROVIDER_ERROR" in response.text
assert '"ok": false' in response.text
def test_tts_failure_keeps_lesson_payload(client):
session_id = create_lesson(client)
app.state.tts = FailingTTS()
response = client.post(
f"/api/v1/sessions/{session_id}/teach",
json={"text": "Teach me."},
)
assert "event: lesson_payload" in response.text
assert "TTS_GENERATION_FAILED" in response.text
assert 'event: done' in response.text
def test_live_artifact_failure_returns_fallback_payload(client):
session_id = create_lesson(client)
app.state.llm = FailingArtifactLLM()
with client.websocket_connect(
f"/api/v1/sessions/{session_id}/live"
) as websocket:
assert websocket.receive_json()["type"] == "ready"
websocket.send_bytes(b"fake-microphone-pcm")
assert websocket.receive_json()["type"] == "input_transcription"
assert websocket.receive_json()["type"] == "output_transcription"
assert websocket.receive_bytes() == b"fake-live-pcm"
lesson_payload = websocket.receive_json()
turn_complete = websocket.receive_json()
assert lesson_payload["type"] == "lesson_payload"
assert lesson_payload["fallback"] is True
assert len(lesson_payload["quiz"]) == 3
assert turn_complete["type"] == "turn_complete"
|