diff --git a/litellm/llms/gigachat/passthrough/transformation.py b/litellm/llms/gigachat/passthrough/transformation.py index d2ba43b0f47..54277081398 100644 --- a/litellm/llms/gigachat/passthrough/transformation.py +++ b/litellm/llms/gigachat/passthrough/transformation.py @@ -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 diff --git a/tests/litellm/llms/gigachat/test_authenticator.py b/tests/litellm/llms/gigachat/test_authenticator.py new file mode 100644 index 00000000000..ed52ae986f5 --- /dev/null +++ b/tests/litellm/llms/gigachat/test_authenticator.py @@ -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") diff --git a/tests/litellm/llms/gigachat/test_utils.py b/tests/litellm/llms/gigachat/test_utils.py new file mode 100644 index 00000000000..00faf391078 --- /dev/null +++ b/tests/litellm/llms/gigachat/test_utils.py @@ -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 diff --git a/tests/test_litellm/llms/gigachat/passthrough/__init__.py b/tests/test_litellm/llms/gigachat/passthrough/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/test_litellm/llms/gigachat/passthrough/test_gigachat_passthrough_transformation.py b/tests/test_litellm/llms/gigachat/passthrough/test_gigachat_passthrough_transformation.py new file mode 100644 index 00000000000..b3e1500c4a3 --- /dev/null +++ b/tests/test_litellm/llms/gigachat/passthrough/test_gigachat_passthrough_transformation.py @@ -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 == []