| from __future__ import annotations | |
| def test_ask_stream_invokes_provider_once(client, monkeypatch) -> None: | |
| from app.routes import ask | |
| class CountingProvider: | |
| model_name = "counting-provider" | |
| is_fallback = False | |
| def __init__(self) -> None: | |
| self.streaming_calls = 0 | |
| def generate_streaming(self, **_kwargs): | |
| self.streaming_calls += 1 | |
| yield "A fast " | |
| yield "answer." | |
| provider = CountingProvider() | |
| monkeypatch.setattr(ask, "get_ai_provider", lambda: provider) | |
| response = client.post( | |
| "/ask/stream", | |
| json={"question": "Explain resonance simply.", "language_preference": "English"}, | |
| ) | |
| assert response.status_code == 200 | |
| assert "A fast " in response.text | |
| assert "answer." in response.text | |
| assert provider.streaming_calls == 1 | |