mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-11 22:51:28 +00:00
Add gigachat tests
This commit is contained in:
KnyazSh
parent
8599009b28
commit
0cc9147f49
5 changed files with 1001 additions and 6 deletions
|
|
@ -1,3 +1,5 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING, cast
|
||||
|
||||
|
|
@ -30,7 +32,7 @@ class GigaChatPassthroughConfig(BasePassthroughConfig):
|
|||
endpoint: str,
|
||||
request_query_params: dict | None,
|
||||
litellm_params: dict,
|
||||
) -> tuple["URL", str]:
|
||||
) -> tuple[URL, str]:
|
||||
"""Get complete API URL for chat completions."""
|
||||
base_target_url = self.get_api_base(api_base)
|
||||
|
||||
|
|
@ -72,11 +74,11 @@ class GigaChatPassthroughConfig(BasePassthroughConfig):
|
|||
self,
|
||||
model: str,
|
||||
custom_llm_provider: str,
|
||||
httpx_response: "Response",
|
||||
httpx_response: Response,
|
||||
request_data: dict,
|
||||
logging_obj: "LiteLLMLoggingObj",
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
endpoint: str,
|
||||
) -> "CostResponseTypes" | None:
|
||||
) -> CostResponseTypes | None:
|
||||
from litellm import encoding
|
||||
from litellm.types.utils import LlmProviders, ModelResponse
|
||||
from litellm.utils import ProviderConfigManager
|
||||
|
|
@ -141,11 +143,11 @@ class GigaChatPassthroughConfig(BasePassthroughConfig):
|
|||
def handle_logging_collected_chunks(
|
||||
self,
|
||||
all_chunks: list[str],
|
||||
litellm_logging_obj: "LiteLLMLoggingObj",
|
||||
litellm_logging_obj: LiteLLMLoggingObj,
|
||||
model: str,
|
||||
custom_llm_provider: str,
|
||||
endpoint: str,
|
||||
) -> "CostResponseTypes" | None:
|
||||
) -> CostResponseTypes | None:
|
||||
"""
|
||||
1. Convert all_chunks to a ModelResponseStream
|
||||
2. combine model_response_stream to model_response
|
||||
|
|
|
|||
428
tests/litellm/llms/gigachat/test_authenticator.py
Normal file
428
tests/litellm/llms/gigachat/test_authenticator.py
Normal file
|
|
@ -0,0 +1,428 @@
|
|||
"""
|
||||
Tests for litellm.llms.gigachat.authenticator
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../../../../../"))
|
||||
|
||||
from litellm.llms.gigachat.authenticator import (
|
||||
GIGACHAT_AUTH_URL,
|
||||
GIGACHAT_SCOPE,
|
||||
GigaChatAuthError,
|
||||
_get_auth_url,
|
||||
_get_credentials,
|
||||
_get_scope,
|
||||
_parse_token_response,
|
||||
_request_token_async,
|
||||
_request_token_sync,
|
||||
get_access_token,
|
||||
get_access_token_async,
|
||||
)
|
||||
|
||||
|
||||
class TestParseTokenResponse:
|
||||
def test_parse_with_tok_and_exp(self):
|
||||
response = MagicMock()
|
||||
response.json.return_value = {"tok": "token123", "exp": 1234567890000}
|
||||
token, expires_at = _parse_token_response(response)
|
||||
assert token == "token123"
|
||||
assert expires_at == 1234567890000
|
||||
|
||||
def test_parse_with_access_token_and_expires_at(self):
|
||||
response = MagicMock()
|
||||
response.json.return_value = {
|
||||
"access_token": "token456",
|
||||
"expires_at": 9876543210000,
|
||||
}
|
||||
token, expires_at = _parse_token_response(response)
|
||||
assert token == "token456"
|
||||
assert expires_at == 9876543210000
|
||||
|
||||
def test_parse_with_string_expires_at(self):
|
||||
response = MagicMock()
|
||||
response.json.return_value = {
|
||||
"access_token": "token789",
|
||||
"expires_at": "1234567890000",
|
||||
}
|
||||
token, expires_at = _parse_token_response(response)
|
||||
assert token == "token789"
|
||||
assert expires_at == 1234567890000
|
||||
|
||||
def test_parse_prefers_tok_over_access_token(self):
|
||||
response = MagicMock()
|
||||
response.json.return_value = {
|
||||
"tok": "preferred",
|
||||
"access_token": "fallback",
|
||||
"exp": 111111,
|
||||
}
|
||||
token, expires_at = _parse_token_response(response)
|
||||
assert token == "preferred"
|
||||
assert expires_at == 111111
|
||||
|
||||
def test_parse_missing_token_raises(self):
|
||||
response = MagicMock()
|
||||
response.json.return_value = {"expires_at": 1234567890000}
|
||||
with pytest.raises(GigaChatAuthError) as exc_info:
|
||||
_parse_token_response(response)
|
||||
assert "Invalid token response" in str(exc_info.value)
|
||||
assert exc_info.value.status_code == 500
|
||||
|
||||
|
||||
class TestGetCredentials:
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_credentials_from_gigachat_credentials(self, mock_get_secret):
|
||||
mock_get_secret.side_effect = lambda key: "cred123" if key == "GIGACHAT_CREDENTIALS" else None
|
||||
result = _get_credentials()
|
||||
assert result == "cred123"
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_credentials_fallback_to_api_key(self, mock_get_secret):
|
||||
mock_get_secret.side_effect = lambda key: (
|
||||
"apikey456" if key == "GIGACHAT_API_KEY" else None
|
||||
)
|
||||
result = _get_credentials()
|
||||
assert result == "apikey456"
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_credentials_returns_none(self, mock_get_secret):
|
||||
mock_get_secret.return_value = None
|
||||
result = _get_credentials()
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestGetAuthUrl:
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_auth_url_from_env(self, mock_get_secret):
|
||||
mock_get_secret.return_value = "https://custom.auth.url"
|
||||
result = _get_auth_url()
|
||||
assert result == "https://custom.auth.url"
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_auth_url_default(self, mock_get_secret):
|
||||
mock_get_secret.return_value = None
|
||||
result = _get_auth_url()
|
||||
assert result == GIGACHAT_AUTH_URL
|
||||
|
||||
|
||||
class TestGetScope:
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_scope_from_env(self, mock_get_secret):
|
||||
mock_get_secret.return_value = "CUSTOM_SCOPE"
|
||||
result = _get_scope()
|
||||
assert result == "CUSTOM_SCOPE"
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_scope_default(self, mock_get_secret):
|
||||
mock_get_secret.return_value = None
|
||||
result = _get_scope()
|
||||
assert result == GIGACHAT_SCOPE
|
||||
|
||||
|
||||
class TestRequestTokenSync:
|
||||
@patch("litellm.llms.gigachat.authenticator.uuid.uuid4")
|
||||
@patch("litellm.llms.gigachat.authenticator._get_http_client")
|
||||
def test_request_token_success(self, mock_get_client, mock_uuid):
|
||||
mock_uuid.return_value = "test-uuid-123"
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {"tok": "newtoken", "exp": 9999999999999}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_client = MagicMock()
|
||||
mock_client.post.return_value = mock_response
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
token, expires_at = _request_token_sync("creds", "SCOPE", "https://auth.url")
|
||||
|
||||
assert token == "newtoken"
|
||||
assert expires_at == 9999999999999
|
||||
mock_client.post.assert_called_once_with(
|
||||
"https://auth.url",
|
||||
headers={
|
||||
"Authorization": "Basic creds",
|
||||
"RqUID": "test-uuid-123",
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
},
|
||||
data={"scope": "SCOPE"},
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator._get_http_client")
|
||||
def test_request_token_http_status_error(self, mock_get_client):
|
||||
mock_response = MagicMock()
|
||||
mock_response.text = "Unauthorized"
|
||||
mock_response.status_code = 401
|
||||
mock_client = MagicMock()
|
||||
mock_client.post.return_value = mock_response
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
|
||||
"401 Unauthorized",
|
||||
request=MagicMock(),
|
||||
response=mock_response,
|
||||
)
|
||||
|
||||
with pytest.raises(GigaChatAuthError) as exc_info:
|
||||
_request_token_sync("creds", "SCOPE", "https://auth.url")
|
||||
assert exc_info.value.status_code == 401
|
||||
assert "Unauthorized" in str(exc_info.value)
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator._get_http_client")
|
||||
def test_request_token_request_error(self, mock_get_client):
|
||||
mock_client = MagicMock()
|
||||
mock_client.post.side_effect = httpx.RequestError("Connection refused")
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
with pytest.raises(GigaChatAuthError) as exc_info:
|
||||
_request_token_sync("creds", "SCOPE", "https://auth.url")
|
||||
assert exc_info.value.status_code == 500
|
||||
assert "Connection refused" in str(exc_info.value)
|
||||
|
||||
|
||||
class TestRequestTokenAsync:
|
||||
@patch("litellm.llms.gigachat.authenticator.uuid.uuid4")
|
||||
@patch("litellm.llms.gigachat.authenticator.get_async_httpx_client")
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_token_async_success(self, mock_get_client, mock_uuid):
|
||||
mock_uuid.return_value = "test-uuid-456"
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {"tok": "async_token", "exp": 8888888888888}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.return_value = mock_response
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
token, expires_at = await _request_token_async("creds", "SCOPE", "https://auth.url")
|
||||
|
||||
assert token == "async_token"
|
||||
assert expires_at == 8888888888888
|
||||
mock_client.post.assert_awaited_once_with(
|
||||
"https://auth.url",
|
||||
headers={
|
||||
"Authorization": "Basic creds",
|
||||
"RqUID": "test-uuid-456",
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
},
|
||||
data={"scope": "SCOPE"},
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator.get_async_httpx_client")
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_token_async_http_status_error(self, mock_get_client):
|
||||
mock_response = MagicMock()
|
||||
mock_response.text = "Forbidden"
|
||||
mock_response.status_code = 403
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.return_value = mock_response
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
|
||||
"403 Forbidden",
|
||||
request=MagicMock(),
|
||||
response=mock_response,
|
||||
)
|
||||
|
||||
with pytest.raises(GigaChatAuthError) as exc_info:
|
||||
await _request_token_async("creds", "SCOPE", "https://auth.url")
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "Forbidden" in str(exc_info.value)
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator.get_async_httpx_client")
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_token_async_request_error(self, mock_get_client):
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.side_effect = httpx.RequestError("Timeout")
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
with pytest.raises(GigaChatAuthError) as exc_info:
|
||||
await _request_token_async("creds", "SCOPE", "https://auth.url")
|
||||
assert exc_info.value.status_code == 500
|
||||
assert "Timeout" in str(exc_info.value)
|
||||
|
||||
|
||||
class TestGetAccessToken:
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_access_token_from_litellm_params(self, mock_get_secret):
|
||||
result = get_access_token(
|
||||
credentials=None,
|
||||
litellm_params={"gigachat_access_token": "param_token"},
|
||||
)
|
||||
assert result == "param_token"
|
||||
mock_get_secret.assert_not_called()
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_access_token_from_env(self, mock_get_secret):
|
||||
mock_get_secret.return_value = "env_token"
|
||||
result = get_access_token(
|
||||
credentials=None,
|
||||
litellm_params={},
|
||||
)
|
||||
assert result == "env_token"
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator._request_token_sync")
|
||||
@patch("litellm.llms.gigachat.authenticator._token_cache")
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_access_token_from_cache_valid(self, mock_get_secret, mock_cache, mock_request):
|
||||
mock_get_secret.side_effect = lambda key: "creds" if key == "GIGACHAT_CREDENTIALS" else None
|
||||
mock_cache.get_cache.return_value = ("cached_token", 9999999999999)
|
||||
|
||||
with patch("time.time", return_value=1000):
|
||||
result = get_access_token(credentials="creds", litellm_params={})
|
||||
|
||||
assert result == "cached_token"
|
||||
mock_request.assert_not_called()
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator._request_token_sync")
|
||||
@patch("litellm.llms.gigachat.authenticator._token_cache")
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_access_token_from_cache_expired(self, mock_get_secret, mock_cache, mock_request):
|
||||
mock_get_secret.side_effect = lambda key: "creds" if key == "GIGACHAT_CREDENTIALS" else None
|
||||
# token expired: 1,050,000 - 60,000 = 990,000 <= 1,000,000
|
||||
mock_cache.get_cache.return_value = ("expired_token", 1050000)
|
||||
mock_request.return_value = ("new_token", 2000000)
|
||||
|
||||
with patch("time.time", return_value=1000):
|
||||
result = get_access_token(credentials="creds", litellm_params={})
|
||||
|
||||
assert result == "new_token"
|
||||
mock_request.assert_called_once()
|
||||
mock_cache.set_cache.assert_called_once()
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator._request_token_sync")
|
||||
@patch("litellm.llms.gigachat.authenticator._token_cache")
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_access_token_requests_new_and_caches(self, mock_get_secret, mock_cache, mock_request):
|
||||
mock_get_secret.side_effect = lambda key: "creds" if key == "GIGACHAT_CREDENTIALS" else None
|
||||
mock_cache.get_cache.return_value = None
|
||||
mock_request.return_value = ("fresh_token", 9999999999999)
|
||||
|
||||
with patch("time.time", return_value=1000):
|
||||
result = get_access_token(credentials="creds", litellm_params={})
|
||||
|
||||
assert result == "fresh_token"
|
||||
mock_request.assert_called_once_with("creds", GIGACHAT_SCOPE, GIGACHAT_AUTH_URL)
|
||||
mock_cache.set_cache.assert_called_once()
|
||||
# check cache key includes first 16 chars of credentials
|
||||
args, kwargs = mock_cache.set_cache.call_args
|
||||
assert args[0] == "gigachat_token:creds"
|
||||
assert args[1] == ("fresh_token", 9999999999999)
|
||||
|
||||
def test_get_access_token_no_credentials_raises(self):
|
||||
with patch("litellm.llms.gigachat.authenticator.get_secret_str", return_value=None):
|
||||
with pytest.raises(GigaChatAuthError) as exc_info:
|
||||
get_access_token(credentials=None, litellm_params={})
|
||||
assert exc_info.value.status_code == 401
|
||||
assert "credentials not provided" in str(exc_info.value)
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_access_token_custom_scope_and_auth_url(self, mock_get_secret):
|
||||
mock_get_secret.return_value = None
|
||||
with patch("litellm.llms.gigachat.authenticator._request_token_sync") as mock_request:
|
||||
mock_request.return_value = ("token", 9999999999999)
|
||||
with patch("litellm.llms.gigachat.authenticator._token_cache") as mock_cache:
|
||||
mock_cache.get_cache.return_value = None
|
||||
with patch("time.time", return_value=1000):
|
||||
result = get_access_token(
|
||||
credentials="creds",
|
||||
scope="CUSTOM_SCOPE",
|
||||
auth_url="https://custom.auth",
|
||||
litellm_params={},
|
||||
)
|
||||
assert result == "token"
|
||||
mock_request.assert_called_once_with("creds", "CUSTOM_SCOPE", "https://custom.auth")
|
||||
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
def test_get_access_token_scope_from_litellm_params(self, mock_get_secret):
|
||||
mock_get_secret.return_value = None
|
||||
with patch("litellm.llms.gigachat.authenticator._request_token_sync") as mock_request:
|
||||
mock_request.return_value = ("token", 9999999999999)
|
||||
with patch("litellm.llms.gigachat.authenticator._token_cache") as mock_cache:
|
||||
mock_cache.get_cache.return_value = None
|
||||
with patch("time.time", return_value=1000):
|
||||
result = get_access_token(
|
||||
credentials="creds",
|
||||
litellm_params={"gigachat_scope": "PARAM_SCOPE", "gigachat_auth_url": "https://param.auth"},
|
||||
)
|
||||
assert result == "token"
|
||||
mock_request.assert_called_once_with("creds", "PARAM_SCOPE", "https://param.auth")
|
||||
|
||||
|
||||
class TestGetAccessTokenAsync:
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_access_token_async_from_litellm_params(self):
|
||||
result = await get_access_token_async(
|
||||
credentials=None,
|
||||
litellm_params={"gigachat_access_token": "async_param_token"},
|
||||
)
|
||||
assert result == "async_param_token"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
async def test_get_access_token_async_from_env(self, mock_get_secret):
|
||||
mock_get_secret.return_value = "async_env_token"
|
||||
result = await get_access_token_async(
|
||||
credentials=None,
|
||||
litellm_params={},
|
||||
)
|
||||
assert result == "async_env_token"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("litellm.llms.gigachat.authenticator._request_token_async")
|
||||
@patch("litellm.llms.gigachat.authenticator._token_cache")
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
async def test_get_access_token_async_from_cache_valid(self, mock_get_secret, mock_cache, mock_request):
|
||||
mock_get_secret.side_effect = lambda key: "creds" if key == "GIGACHAT_CREDENTIALS" else None
|
||||
mock_cache.get_cache.return_value = ("cached_async_token", 9999999999999)
|
||||
|
||||
with patch("time.time", return_value=1000):
|
||||
result = await get_access_token_async(credentials="creds", litellm_params={})
|
||||
|
||||
assert result == "cached_async_token"
|
||||
mock_request.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("litellm.llms.gigachat.authenticator._request_token_async")
|
||||
@patch("litellm.llms.gigachat.authenticator._token_cache")
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
async def test_get_access_token_async_requests_new_and_caches(self, mock_get_secret, mock_cache, mock_request):
|
||||
mock_get_secret.side_effect = lambda key: "creds" if key == "GIGACHAT_CREDENTIALS" else None
|
||||
mock_cache.get_cache.return_value = None
|
||||
mock_request.return_value = ("fresh_async_token", 9999999999999)
|
||||
|
||||
with patch("time.time", return_value=1000):
|
||||
result = await get_access_token_async(credentials="creds", litellm_params={})
|
||||
|
||||
assert result == "fresh_async_token"
|
||||
mock_request.assert_awaited_once_with("creds", GIGACHAT_SCOPE, GIGACHAT_AUTH_URL)
|
||||
mock_cache.set_cache.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_access_token_async_no_credentials_raises(self):
|
||||
with patch("litellm.llms.gigachat.authenticator.get_secret_str", return_value=None):
|
||||
with pytest.raises(GigaChatAuthError) as exc_info:
|
||||
await get_access_token_async(credentials=None, litellm_params={})
|
||||
assert exc_info.value.status_code == 401
|
||||
assert "credentials not provided" in str(exc_info.value)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("litellm.llms.gigachat.authenticator.get_secret_str")
|
||||
async def test_get_access_token_async_custom_params(self, mock_get_secret):
|
||||
mock_get_secret.return_value = None
|
||||
with patch("litellm.llms.gigachat.authenticator._request_token_async") as mock_request:
|
||||
mock_request.return_value = ("token", 9999999999999)
|
||||
with patch("litellm.llms.gigachat.authenticator._token_cache") as mock_cache:
|
||||
mock_cache.get_cache.return_value = None
|
||||
with patch("time.time", return_value=1000):
|
||||
result = await get_access_token_async(
|
||||
credentials="creds",
|
||||
scope="CUSTOM",
|
||||
auth_url="https://custom",
|
||||
litellm_params={},
|
||||
)
|
||||
assert result == "token"
|
||||
mock_request.assert_awaited_once_with("creds", "CUSTOM", "https://custom")
|
||||
84
tests/litellm/llms/gigachat/test_utils.py
Normal file
84
tests/litellm/llms/gigachat/test_utils.py
Normal file
|
|
@ -0,0 +1,84 @@
|
|||
"""
|
||||
Tests for litellm.llms.gigachat.utils
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(
|
||||
0, os.path.abspath("../../../../../")
|
||||
) # Adds the project root to the system path
|
||||
|
||||
import pytest
|
||||
from litellm.llms.gigachat.utils import convert_usage
|
||||
from litellm.types.utils import PromptTokensDetailsWrapper, Usage
|
||||
|
||||
|
||||
class TestConvertUsage:
|
||||
def test_basic_usage_without_precached(self):
|
||||
"""Test convert_usage with standard tokens, no precached prompt tokens."""
|
||||
result = convert_usage(
|
||||
{
|
||||
"prompt_tokens": 10,
|
||||
"completion_tokens": 5,
|
||||
"total_tokens": 15,
|
||||
}
|
||||
)
|
||||
|
||||
assert result == Usage(
|
||||
prompt_tokens=10,
|
||||
completion_tokens=5,
|
||||
total_tokens=15,
|
||||
prompt_tokens_details=None,
|
||||
)
|
||||
|
||||
def test_usage_with_precached_prompt_tokens(self):
|
||||
"""Test convert_usage adds precached_prompt_tokens to prompt_tokens and total_tokens."""
|
||||
result = convert_usage(
|
||||
{
|
||||
"prompt_tokens": 10,
|
||||
"completion_tokens": 5,
|
||||
"precached_prompt_tokens": 3,
|
||||
"total_tokens": 15,
|
||||
}
|
||||
)
|
||||
|
||||
assert result == Usage(
|
||||
prompt_tokens=13,
|
||||
completion_tokens=5,
|
||||
total_tokens=18,
|
||||
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=3),
|
||||
)
|
||||
|
||||
def test_zero_precached_prompt_tokens(self):
|
||||
"""Test convert_usage with zero precached_prompt_tokens does not create details wrapper."""
|
||||
result = convert_usage(
|
||||
{
|
||||
"prompt_tokens": 10,
|
||||
"completion_tokens": 5,
|
||||
"precached_prompt_tokens": 0,
|
||||
"total_tokens": 15,
|
||||
}
|
||||
)
|
||||
|
||||
assert result == Usage(
|
||||
prompt_tokens=10,
|
||||
completion_tokens=5,
|
||||
total_tokens=15,
|
||||
prompt_tokens_details=None,
|
||||
)
|
||||
|
||||
def test_missing_optional_fields(self):
|
||||
"""Test convert_usage with missing optional fields defaults to zero."""
|
||||
result = convert_usage(
|
||||
{
|
||||
"prompt_tokens": 10,
|
||||
"completion_tokens": 5,
|
||||
"total_tokens": 15,
|
||||
}
|
||||
)
|
||||
|
||||
assert result.prompt_tokens == 10
|
||||
assert result.completion_tokens == 5
|
||||
assert result.total_tokens == 15
|
||||
assert result.prompt_tokens_details is None
|
||||
0
tests/test_litellm/llms/gigachat/passthrough/__init__.py
Normal file
0
tests/test_litellm/llms/gigachat/passthrough/__init__.py
Normal file
|
|
@ -0,0 +1,481 @@
|
|||
"""
|
||||
Unit tests for GigaChatPassthroughConfig transformation.
|
||||
|
||||
Tests the GigaChat-specific passthrough configuration including URL construction,
|
||||
streaming detection, authentication handling, and logging response transformations.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../../../../.."))
|
||||
|
||||
from litellm.llms.gigachat.passthrough.transformation import GigaChatPassthroughConfig
|
||||
from litellm.types.utils import EmbeddingResponse, ModelResponse
|
||||
|
||||
|
||||
def _gigachat_chat_completion_body():
|
||||
return {
|
||||
"id": "chatcmpl-test123",
|
||||
"object": "chat.completion",
|
||||
"created": 1700000000,
|
||||
"model": "GigaChat",
|
||||
"choices": [
|
||||
{
|
||||
"index": 0,
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": "Hello from GigaChat",
|
||||
},
|
||||
"finish_reason": "stop",
|
||||
}
|
||||
],
|
||||
"usage": {
|
||||
"prompt_tokens": 5,
|
||||
"completion_tokens": 3,
|
||||
"total_tokens": 8,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _gigachat_embedding_body():
|
||||
return {
|
||||
"object": "list",
|
||||
"data": [
|
||||
{
|
||||
"object": "embedding",
|
||||
"embedding": [0.1, 0.2, 0.3],
|
||||
"index": 0,
|
||||
"usage": {"prompt_tokens": 4},
|
||||
}
|
||||
],
|
||||
"model": "Embeddings",
|
||||
}
|
||||
|
||||
|
||||
def _make_httpx_response(body: dict) -> httpx.Response:
|
||||
return httpx.Response(
|
||||
status_code=200,
|
||||
headers={"content-type": "application/json"},
|
||||
content=json.dumps(body).encode("utf-8"),
|
||||
request=httpx.Request(
|
||||
"POST", "https://gigachat.devices.sberbank.ru/api/v1/chat/completions"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class TestGigaChatPassthroughConfig:
|
||||
"""Tests for GigaChatPassthroughConfig class."""
|
||||
|
||||
def test_is_streaming_request_true(self):
|
||||
"""Test streaming is detected when stream=True."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
assert (
|
||||
config.is_streaming_request("chat/completions", {"stream": True}) is True
|
||||
)
|
||||
|
||||
def test_is_streaming_request_false(self):
|
||||
"""Test streaming is not detected when stream=False."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
assert (
|
||||
config.is_streaming_request("chat/completions", {"stream": False})
|
||||
is False
|
||||
)
|
||||
|
||||
def test_is_streaming_request_missing_stream_key(self):
|
||||
"""Test streaming defaults to False when stream key is missing."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
assert (
|
||||
config.is_streaming_request("chat/completions", {"model": "GigaChat"})
|
||||
is False
|
||||
)
|
||||
|
||||
def test_get_complete_url_with_api_base(self):
|
||||
"""Test URL construction with explicit api_base."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
api_base = "https://custom.gigachat.ru/api/v1"
|
||||
endpoint = "chat/completions"
|
||||
|
||||
complete_url, base_target_url = config.get_complete_url(
|
||||
api_base=api_base,
|
||||
api_key=None,
|
||||
model="GigaChat",
|
||||
endpoint=endpoint,
|
||||
request_query_params=None,
|
||||
litellm_params={},
|
||||
)
|
||||
|
||||
assert isinstance(complete_url, httpx.URL)
|
||||
assert str(complete_url) == f"{api_base}/{endpoint}"
|
||||
assert base_target_url == api_base
|
||||
|
||||
def test_get_complete_url_with_leading_slash_endpoint(self):
|
||||
"""Test URL construction with endpoint having leading slash."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
api_base = "https://custom.gigachat.ru/api/v1"
|
||||
endpoint = "/chat/completions"
|
||||
|
||||
complete_url, base_target_url = config.get_complete_url(
|
||||
api_base=api_base,
|
||||
api_key=None,
|
||||
model="GigaChat",
|
||||
endpoint=endpoint,
|
||||
request_query_params=None,
|
||||
litellm_params={},
|
||||
)
|
||||
|
||||
assert str(complete_url) == "https://custom.gigachat.ru/api/v1/chat/completions"
|
||||
assert base_target_url == api_base
|
||||
|
||||
@patch(
|
||||
"litellm.llms.gigachat.passthrough.transformation.get_secret_str"
|
||||
)
|
||||
def test_get_complete_url_with_env_api_base(self, mock_get_secret):
|
||||
"""Test URL construction with api_base from environment."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
env_api_base = "https://env.gigachat.ru/api/v1"
|
||||
mock_get_secret.return_value = env_api_base
|
||||
|
||||
complete_url, base_target_url = config.get_complete_url(
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
model="GigaChat",
|
||||
endpoint="embeddings",
|
||||
request_query_params=None,
|
||||
litellm_params={},
|
||||
)
|
||||
|
||||
assert isinstance(complete_url, httpx.URL)
|
||||
assert str(complete_url).startswith(env_api_base)
|
||||
assert base_target_url == env_api_base
|
||||
mock_get_secret.assert_called_once_with("GIGACHAT_API_BASE")
|
||||
|
||||
@patch(
|
||||
"litellm.llms.gigachat.passthrough.transformation.get_secret_str"
|
||||
)
|
||||
def test_get_complete_url_fallback_to_default(self, mock_get_secret):
|
||||
"""Test URL construction falls back to default GIGACHAT_BASE_URL."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
mock_get_secret.return_value = None
|
||||
|
||||
complete_url, base_target_url = config.get_complete_url(
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
model="GigaChat",
|
||||
endpoint="models",
|
||||
request_query_params=None,
|
||||
litellm_params={},
|
||||
)
|
||||
|
||||
assert isinstance(complete_url, httpx.URL)
|
||||
assert "gigachat.devices.sberbank.ru" in str(complete_url)
|
||||
assert base_target_url == "https://gigachat.devices.sberbank.ru/api/v1"
|
||||
|
||||
def test_get_complete_url_no_api_base_raises(self):
|
||||
"""Test that exception is raised when no api_base can be resolved."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
with patch(
|
||||
"litellm.llms.gigachat.passthrough.transformation.get_secret_str",
|
||||
return_value=None,
|
||||
):
|
||||
with patch(
|
||||
"litellm.llms.gigachat.passthrough.transformation.GIGACHAT_BASE_URL",
|
||||
None,
|
||||
):
|
||||
with pytest.raises(Exception, match="GigaChat api base not found"):
|
||||
config.get_complete_url(
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
model="GigaChat",
|
||||
endpoint="chat/completions",
|
||||
request_query_params=None,
|
||||
litellm_params={},
|
||||
)
|
||||
|
||||
@patch(
|
||||
"litellm.llms.gigachat.passthrough.transformation.get_access_token"
|
||||
)
|
||||
def test_validate_environment(self, mock_get_access_token):
|
||||
"""Test headers are set correctly with OAuth token."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
mock_get_access_token.return_value = "test-token-123"
|
||||
|
||||
headers = config.validate_environment(
|
||||
headers={},
|
||||
model="GigaChat",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
api_key="test-credentials",
|
||||
api_base="https://custom.gigachat.ru",
|
||||
)
|
||||
|
||||
assert headers["Authorization"] == "Bearer test-token-123"
|
||||
assert headers["Content-Type"] == "application/json"
|
||||
assert headers["Accept"] == "application/json"
|
||||
mock_get_access_token.assert_called_once_with(
|
||||
credentials="test-credentials",
|
||||
litellm_params={},
|
||||
)
|
||||
|
||||
def test_logging_non_streaming_response_chat_completions(self):
|
||||
"""Test chat completions endpoint returns ModelResponse."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
logging_obj = MagicMock()
|
||||
|
||||
result = config.logging_non_streaming_response(
|
||||
model="gigachat/GigaChat",
|
||||
custom_llm_provider="gigachat",
|
||||
httpx_response=_make_httpx_response(_gigachat_chat_completion_body()),
|
||||
request_data={
|
||||
"model": "gigachat/GigaChat",
|
||||
"messages": [{"role": "user", "content": "hi"}],
|
||||
},
|
||||
logging_obj=logging_obj,
|
||||
endpoint="chat/completions",
|
||||
)
|
||||
|
||||
assert isinstance(result, ModelResponse)
|
||||
assert result.choices[0].message.content == "Hello from GigaChat"
|
||||
assert result.usage.prompt_tokens == 5
|
||||
assert result.usage.completion_tokens == 3
|
||||
assert result.usage.total_tokens == 8
|
||||
|
||||
def test_logging_non_streaming_response_embeddings(self):
|
||||
"""Test embeddings endpoint returns EmbeddingResponse."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
logging_obj = MagicMock()
|
||||
|
||||
result = config.logging_non_streaming_response(
|
||||
model="gigachat/Embeddings",
|
||||
custom_llm_provider="gigachat",
|
||||
httpx_response=_make_httpx_response(_gigachat_embedding_body()),
|
||||
request_data={"input": ["hello"], "model": "gigachat/Embeddings"},
|
||||
logging_obj=logging_obj,
|
||||
endpoint="embeddings",
|
||||
)
|
||||
|
||||
assert isinstance(result, EmbeddingResponse)
|
||||
assert len(result.data) == 1
|
||||
assert result.data[0]["embedding"] == [0.1, 0.2, 0.3]
|
||||
|
||||
def test_logging_non_streaming_response_unknown_endpoint_returns_none(self):
|
||||
"""Test unknown endpoint returns None."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
logging_obj = MagicMock()
|
||||
|
||||
result = config.logging_non_streaming_response(
|
||||
model="gigachat/GigaChat",
|
||||
custom_llm_provider="gigachat",
|
||||
httpx_response=_make_httpx_response(_gigachat_chat_completion_body()),
|
||||
request_data={},
|
||||
logging_obj=logging_obj,
|
||||
endpoint="images/generations",
|
||||
)
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_handle_logging_collected_chunks_with_string_chunks(self):
|
||||
"""Test converting string chunks to model response."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
logging_obj = MagicMock()
|
||||
|
||||
chunks = [
|
||||
'{"choices": [{"delta": {"content": "Hello"}, "index": 0}]}',
|
||||
'{"choices": [{"delta": {"content": " world"}, "index": 0}]}',
|
||||
'{"choices": [{"delta": {}, "finish_reason": "stop", "index": 0}], "usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7}}',
|
||||
]
|
||||
|
||||
result = config.handle_logging_collected_chunks(
|
||||
all_chunks=chunks,
|
||||
litellm_logging_obj=logging_obj,
|
||||
model="gigachat/GigaChat",
|
||||
custom_llm_provider="gigachat",
|
||||
endpoint="chat/completions",
|
||||
)
|
||||
|
||||
assert isinstance(result, ModelResponse)
|
||||
assert result.choices[0].message.content == "Hello world"
|
||||
|
||||
def test_handle_logging_collected_chunks_with_bytes_chunks(self):
|
||||
"""Test converting bytes chunks to model response."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
logging_obj = MagicMock()
|
||||
|
||||
chunks = [
|
||||
b'{"choices": [{"delta": {"content": "Hi"}, "index": 0}]}',
|
||||
b'{"choices": [{"delta": {}, "finish_reason": "stop", "index": 0}], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}}',
|
||||
]
|
||||
|
||||
result = config.handle_logging_collected_chunks(
|
||||
all_chunks=chunks,
|
||||
litellm_logging_obj=logging_obj,
|
||||
model="gigachat/GigaChat",
|
||||
custom_llm_provider="gigachat",
|
||||
endpoint="chat/completions",
|
||||
)
|
||||
|
||||
assert isinstance(result, ModelResponse)
|
||||
assert result.choices[0].message.content == "Hi"
|
||||
|
||||
def test_handle_logging_collected_chunks_with_done_and_empty(self):
|
||||
"""Test that [DONE] and empty chunks are skipped."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
logging_obj = MagicMock()
|
||||
|
||||
chunks = [
|
||||
"",
|
||||
"[DONE]",
|
||||
'{"choices": [{"delta": {"content": "test"}, "index": 0}]}',
|
||||
'{"choices": [{"delta": {}, "finish_reason": "stop", "index": 0}], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}}',
|
||||
]
|
||||
|
||||
result = config.handle_logging_collected_chunks(
|
||||
all_chunks=chunks,
|
||||
litellm_logging_obj=logging_obj,
|
||||
model="gigachat/GigaChat",
|
||||
custom_llm_provider="gigachat",
|
||||
endpoint="chat/completions",
|
||||
)
|
||||
|
||||
assert isinstance(result, ModelResponse)
|
||||
assert result.choices[0].message.content == "test"
|
||||
|
||||
def test_handle_logging_collected_chunks_with_dict_chunks(self):
|
||||
"""Test converting dict chunks directly."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
logging_obj = MagicMock()
|
||||
|
||||
chunks = [
|
||||
{"choices": [{"delta": {"content": "direct"}, "index": 0}]},
|
||||
{
|
||||
"choices": [
|
||||
{
|
||||
"delta": {},
|
||||
"finish_reason": "stop",
|
||||
"index": 0,
|
||||
}
|
||||
],
|
||||
"usage": {
|
||||
"prompt_tokens": 1,
|
||||
"completion_tokens": 1,
|
||||
"total_tokens": 2,
|
||||
},
|
||||
},
|
||||
]
|
||||
|
||||
result = config.handle_logging_collected_chunks(
|
||||
all_chunks=chunks,
|
||||
litellm_logging_obj=logging_obj,
|
||||
model="gigachat/GigaChat",
|
||||
custom_llm_provider="gigachat",
|
||||
endpoint="chat/completions",
|
||||
)
|
||||
|
||||
assert isinstance(result, ModelResponse)
|
||||
assert result.choices[0].message.content == "direct"
|
||||
|
||||
def test_handle_logging_collected_chunks_empty_list_returns_none(self):
|
||||
"""Test empty chunks list returns None."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
logging_obj = MagicMock()
|
||||
|
||||
result = config.handle_logging_collected_chunks(
|
||||
all_chunks=[],
|
||||
litellm_logging_obj=logging_obj,
|
||||
model="gigachat/GigaChat",
|
||||
custom_llm_provider="gigachat",
|
||||
endpoint="chat/completions",
|
||||
)
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_handle_logging_collected_chunks_invalid_json_skipped(self):
|
||||
"""Test invalid JSON chunks are skipped gracefully."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
logging_obj = MagicMock()
|
||||
|
||||
chunks = [
|
||||
"not-valid-json",
|
||||
'{"choices": [{"delta": {"content": "valid"}, "index": 0}]}',
|
||||
'{"choices": [{"delta": {}, "finish_reason": "stop", "index": 0}], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}}',
|
||||
]
|
||||
|
||||
result = config.handle_logging_collected_chunks(
|
||||
all_chunks=chunks,
|
||||
litellm_logging_obj=logging_obj,
|
||||
model="gigachat/GigaChat",
|
||||
custom_llm_provider="gigachat",
|
||||
endpoint="chat/completions",
|
||||
)
|
||||
|
||||
assert isinstance(result, ModelResponse)
|
||||
assert result.choices[0].message.content == "valid"
|
||||
|
||||
@patch(
|
||||
"litellm.llms.gigachat.passthrough.transformation.get_secret_str"
|
||||
)
|
||||
def test_get_api_base_with_explicit_value(self, mock_get_secret):
|
||||
"""Test get_api_base returns explicit value when provided."""
|
||||
explicit_base = "https://custom.gigachat.ru/api/v1"
|
||||
result = GigaChatPassthroughConfig.get_api_base(api_base=explicit_base)
|
||||
assert result == explicit_base
|
||||
mock_get_secret.assert_not_called()
|
||||
|
||||
@patch(
|
||||
"litellm.llms.gigachat.passthrough.transformation.get_secret_str"
|
||||
)
|
||||
def test_get_api_base_from_environment(self, mock_get_secret):
|
||||
"""Test get_api_base retrieves from environment when not provided."""
|
||||
env_base = "https://env.gigachat.ru/api/v1"
|
||||
mock_get_secret.return_value = env_base
|
||||
result = GigaChatPassthroughConfig.get_api_base(api_base=None)
|
||||
assert result == env_base
|
||||
mock_get_secret.assert_called_once_with("GIGACHAT_API_BASE")
|
||||
|
||||
@patch(
|
||||
"litellm.llms.gigachat.passthrough.transformation.get_secret_str"
|
||||
)
|
||||
def test_get_api_base_fallback_to_default(self, mock_get_secret):
|
||||
"""Test get_api_base falls back to GIGACHAT_BASE_URL."""
|
||||
mock_get_secret.return_value = None
|
||||
result = GigaChatPassthroughConfig.get_api_base(api_base=None)
|
||||
assert result == "https://gigachat.devices.sberbank.ru/api/v1"
|
||||
|
||||
@patch(
|
||||
"litellm.llms.gigachat.passthrough.transformation.get_secret_str"
|
||||
)
|
||||
def test_get_api_key_with_explicit_value(self, mock_get_secret):
|
||||
"""Test get_api_key returns explicit value when provided."""
|
||||
explicit_key = "test-api-key"
|
||||
result = GigaChatPassthroughConfig.get_api_key(api_key=explicit_key)
|
||||
assert result == explicit_key
|
||||
mock_get_secret.assert_not_called()
|
||||
|
||||
@patch(
|
||||
"litellm.llms.gigachat.passthrough.transformation.get_secret_str"
|
||||
)
|
||||
def test_get_api_key_from_environment(self, mock_get_secret):
|
||||
"""Test get_api_key retrieves from environment when not provided."""
|
||||
env_key = "env-api-key"
|
||||
mock_get_secret.return_value = env_key
|
||||
result = GigaChatPassthroughConfig.get_api_key(api_key=None)
|
||||
assert result == env_key
|
||||
mock_get_secret.assert_called_once_with("GIGACHAT_API_KEY")
|
||||
|
||||
def test_get_base_model_returns_model(self):
|
||||
"""Test get_base_model returns the model as-is."""
|
||||
model = "gigachat/GigaChat"
|
||||
result = GigaChatPassthroughConfig.get_base_model(model)
|
||||
assert result == model
|
||||
|
||||
def test_get_models(self):
|
||||
"""Test get_models delegates to base class."""
|
||||
config = GigaChatPassthroughConfig()
|
||||
result = config.get_models()
|
||||
assert result == []
|
||||
Loading…
Add table
Reference in a new issue