Spaces:
Running
Running
| import os | |
| from collections.abc import AsyncIterator | |
| from contextlib import asynccontextmanager | |
| from pathlib import Path | |
| from types import SimpleNamespace | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| os.environ["DATABASE_URL"] = "sqlite:///./data/test_gargi_ai.db" | |
| os.environ["GEMINI_API_KEY"] = "test-key" | |
| from gargi_ai.config import get_settings # noqa: E402 | |
| from gargi_ai.database import Base, engine # noqa: E402 | |
| from gargi_ai.main import app # noqa: E402 | |
| from gargi_ai.providers import LiveProvider, LLMProvider, TTSProvider # noqa: E402 | |
| from gargi_ai.schemas import QuizQuestion, TeachingPayload, WhiteboardBlock # noqa: E402 | |
| class FakeLLM(LLMProvider): | |
| async def stream_explanation(self, prompt: str) -> AsyncIterator[str]: | |
| yield "Gravity " | |
| yield "pulls objects together." | |
| async def generate_artifact(self, prompt: str) -> TeachingPayload: | |
| return TeachingPayload( | |
| spoken_explanation="Gravity pulls objects together.", | |
| summary="Gravity is an attractive force.", | |
| whiteboard_blocks=[ | |
| WhiteboardBlock(type="heading", content="Gravity"), | |
| WhiteboardBlock(type="equation", content="F = Gm1m2/r^2"), | |
| ], | |
| quiz=[ | |
| QuizQuestion( | |
| id="q1", | |
| type="multiple_choice", | |
| question="What does gravity do?", | |
| options=["Attracts", "Repels"], | |
| expected_answer="Attracts", | |
| explanation="Masses attract.", | |
| ), | |
| QuizQuestion( | |
| id="q2", | |
| type="multiple_choice", | |
| question="Which has gravity?", | |
| options=["Mass", "Nothing"], | |
| expected_answer="Mass", | |
| explanation="All mass gravitates.", | |
| ), | |
| QuizQuestion( | |
| id="q3", | |
| type="short_answer", | |
| question="Name the force.", | |
| expected_answer="Gravity", | |
| explanation="The force is gravity.", | |
| ), | |
| ], | |
| detected_misconceptions=[], | |
| learner_level_update="beginner", | |
| suggested_follow_up="Explore orbits.", | |
| ) | |
| class FakeTTS(TTSProvider): | |
| def __init__(self, audio_dir: Path): | |
| self.audio_dir = audio_dir | |
| async def synthesize(self, text: str, voice: str): | |
| audio_id = "11111111-1111-1111-1111-111111111111" | |
| output = self.audio_dir / f"{audio_id}.mp3" | |
| output.parent.mkdir(parents=True, exist_ok=True) | |
| output.write_bytes(b"fake-mp3") | |
| return audio_id, output | |
| class FakeLiveSession: | |
| def __init__(self): | |
| self.received_audio = False | |
| self.delivered = False | |
| async def send_realtime_input(self, **kwargs): | |
| if kwargs.get("audio"): | |
| self.received_audio = True | |
| async def receive(self): | |
| while not self.received_audio or self.delivered: | |
| import asyncio | |
| await asyncio.sleep(0.01) | |
| self.delivered = True | |
| part = SimpleNamespace( | |
| inline_data=SimpleNamespace(data=b"fake-live-pcm") | |
| ) | |
| content = SimpleNamespace( | |
| input_transcription=SimpleNamespace(text="Why does gravity work?"), | |
| output_transcription=SimpleNamespace( | |
| text="Gravity attracts objects with mass." | |
| ), | |
| model_turn=SimpleNamespace(parts=[part]), | |
| interrupted=False, | |
| turn_complete=True, | |
| ) | |
| yield SimpleNamespace(server_content=content) | |
| class FakeLive(LiveProvider): | |
| async def connect(self, system_instruction: str): | |
| yield FakeLiveSession() | |
| def client(): | |
| settings = get_settings() | |
| Base.metadata.drop_all(bind=engine) | |
| settings.audio_dir.mkdir(parents=True, exist_ok=True) | |
| with TestClient(app) as test_client: | |
| app.state.llm = FakeLLM() | |
| app.state.tts = FakeTTS(settings.audio_dir) | |
| app.state.live = FakeLive() | |
| yield test_client | |
| Base.metadata.drop_all(bind=engine) | |