49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
import requests
|
|
|
|
from app.modules.shared.gigachat.client import GigaChatClient
|
|
from app.modules.shared.gigachat.settings import GigaChatSettings
|
|
|
|
|
|
class _FakeTokenProvider:
|
|
def get_access_token(self) -> str:
|
|
return "token"
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, status_code: int, payload: dict, text: str = "") -> None:
|
|
self.status_code = status_code
|
|
self._payload = payload
|
|
self.text = text
|
|
|
|
def json(self) -> dict:
|
|
return self._payload
|
|
|
|
|
|
def test_gigachat_client_retries_transient_http_errors(monkeypatch) -> None:
|
|
calls = {"count": 0}
|
|
|
|
def fake_post(*args, **kwargs):
|
|
calls["count"] += 1
|
|
if calls["count"] == 1:
|
|
return _FakeResponse(503, {}, "temporary")
|
|
return _FakeResponse(200, {"choices": [{"message": {"content": "ok"}}]})
|
|
|
|
monkeypatch.setattr(requests, "post", fake_post)
|
|
client = GigaChatClient(
|
|
GigaChatSettings(
|
|
auth_url="https://auth.example.test",
|
|
api_url="https://api.example.test",
|
|
scope="scope",
|
|
credentials="secret",
|
|
ssl_verify=True,
|
|
model="model",
|
|
embedding_model="embed",
|
|
),
|
|
_FakeTokenProvider(),
|
|
)
|
|
|
|
result = client.complete("system", "user")
|
|
|
|
assert result == "ok"
|
|
assert calls["count"] == 2
|