| import pytest | |
| from app.core.services.llm_service import GrokLLMService | |
| from app.core.exceptions import LLMProviderRateLimit | |
| class Fake429Resp: | |
| def __init__(self, text="rate limit", payload=None): | |
| self.status_code = 429 | |
| self.headers = {} | |
| self.text = text | |
| self._payload = payload or { | |
| "code": "Some resource has been exhausted", | |
| "error": "out of credits", | |
| } | |
| def json(self): | |
| return self._payload | |
| class FakeAsyncClient: | |
| def __init__(self, *args, **kwargs): | |
| self._calls = 0 | |
| async def __aenter__(self): | |
| return self | |
| async def __aexit__(self, exc_type, exc, tb): | |
| return False | |
| async def post(self, *args, **kwargs): | |
| self._calls += 1 | |
| return Fake429Resp() | |
| def test_grok_service_raises_rate_limit(monkeypatch): | |
| monkeypatch.setattr("app.core.services.llm_service.AsyncClient", FakeAsyncClient) | |
| svc = GrokLLMService() | |
| svc.api_key = "fake" | |
| svc.base_url = "https://fake" | |
| svc.model_name = "grok-test" | |
| with pytest.raises(LLMProviderRateLimit): | |
| import asyncio | |
| asyncio.run(svc.generate("hi there")) | |