fix(test): add tests for gigachat

tests/test_litellm/llms/gigachat/chat/test_gigachat_chat_transformation.py

@@ -0,0 +1,887 @@
+"""
+Unit tests for GigaChat chat transformation.
+
+Tests GigaChatConfig covering get_complete_url, validate_environment,
+get_supported_openai_params, map_openai_params, _convert_tools_to_functions,
+_map_tool_choice, _transform_messages, transform_request, transform_response,
+get_model_response_iterator, and get_error_class.
+"""
+
+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.chat.transformation import (
+    GigaChatConfig,
+    GigaChatError,
+    is_valid_json,
+)
+from litellm.types.utils import ModelResponse, Usage
+
+TRANSFORM_MODULE = "litellm.llms.gigachat.chat.transformation"
+
+
+def _make_httpx_response(
+    body: dict, status_code: int = 200
+) -> httpx.Response:
+    return httpx.Response(
+        status_code=status_code,
+        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",
+        ),
+    )
+
+
+# ---------------------------------------------------------------------------
+# is_valid_json
+# ---------------------------------------------------------------------------
+
+
+class TestIsValidJson:
+    def test_valid_json_object(self):
+        assert is_valid_json('{"key": "value"}') is True
+
+    def test_valid_json_array(self):
+        assert is_valid_json("[1, 2, 3]") is True
+
+    def test_valid_json_string(self):
+        assert is_valid_json('"hello"') is True
+
+    def test_invalid_json(self):
+        assert is_valid_json("{invalid}") is False
+
+    def test_empty_string(self):
+        assert is_valid_json("") is False
+
+
+# ---------------------------------------------------------------------------
+# GigaChatConfig
+# ---------------------------------------------------------------------------
+
+
+class TestGetCompleteUrl:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    def test_uses_api_base_from_param(self):
+        url = self.config.get_complete_url(
+            api_base="https://custom.example.com",
+            api_key=None,
+            model="GigaChat",
+            optional_params={},
+            litellm_params={},
+            stream=False,
+        )
+        assert url == "https://custom.example.com/chat/completions"
+
+    def test_uses_api_base_with_trailing_slash(self):
+        url = self.config.get_complete_url(
+            api_base="https://custom.example.com/",
+            api_key=None,
+            model="GigaChat",
+            optional_params={},
+            litellm_params={},
+            stream=False,
+        )
+        # get_api_base passes the value through without stripping the slash
+        assert url == "https://custom.example.com//chat/completions"
+
+    def test_uses_api_base_from_get_api_base_when_none(self):
+        url = self.config.get_complete_url(
+            api_base=None,
+            api_key=None,
+            model="GigaChat",
+            optional_params={},
+            litellm_params={},
+            stream=False,
+        )
+        assert url.endswith("/chat/completions")
+
+
+class TestValidateEnvironment:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    @patch(f"{TRANSFORM_MODULE}.get_access_token", return_value="test-token")
+    @patch(f"{TRANSFORM_MODULE}.get_secret_str", return_value=None)
+    def test_sets_auth_headers(self, mock_get_secret, mock_get_token):
+        headers: dict = {}
+        result = self.config.validate_environment(
+            headers=headers,
+            model="GigaChat",
+            messages=[{"role": "user", "content": "hi"}],
+            optional_params={},
+            litellm_params={},
+            api_key="creds",
+            api_base="https://api.example.com",
+        )
+        assert result["Authorization"] == "Bearer test-token"
+        assert result["Content-Type"] == "application/json"
+        assert result["Accept"] == "application/json"
+
+    @patch(f"{TRANSFORM_MODULE}.get_access_token", return_value="token")
+    @patch(f"{TRANSFORM_MODULE}.get_secret_str", return_value=None)
+    def test_stores_credentials_and_api_base_for_image_uploads(
+        self, mock_get_secret, mock_get_token
+    ):
+        self.config.validate_environment(
+            headers={},
+            model="GigaChat",
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            api_key="my-creds",
+            api_base="https://my-api.example.com",
+        )
+        assert self.config._current_credentials == "my-creds"
+        assert self.config._current_api_base == "https://my-api.example.com"
+
+    @patch(f"{TRANSFORM_MODULE}.get_access_token", return_value="token")
+    @patch(f"{TRANSFORM_MODULE}.get_secret_str")
+    def test_falls_back_to_env_for_credentials(
+        self, mock_get_secret, mock_get_token
+    ):
+        mock_get_secret.return_value = "env-creds"
+        self.config.validate_environment(
+            headers={},
+            model="GigaChat",
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            api_key=None,
+            api_base=None,
+        )
+        mock_get_secret.assert_any_call("GIGACHAT_CREDENTIALS")
+
+
+class TestGetSupportedOpenAiParams:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    def test_returns_expected_params(self):
+        params = self.config.get_supported_openai_params("GigaChat")
+        expected = [
+            "stream",
+            "temperature",
+            "top_p",
+            "max_tokens",
+            "max_completion_tokens",
+            "stop",
+            "tools",
+            "tool_choice",
+            "functions",
+            "function_call",
+            "response_format",
+        ]
+        assert params == expected
+
+
+class TestMapOpenAiParams:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    def test_stream(self):
+        result = self.config.map_openai_params(
+            non_default_params={"stream": True},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result["stream"] is True
+
+    def test_temperature_zero_maps_to_top_p_zero(self):
+        result = self.config.map_openai_params(
+            non_default_params={"temperature": 0},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result["top_p"] == 0
+        assert "temperature" not in result
+
+    def test_temperature_non_zero(self):
+        result = self.config.map_openai_params(
+            non_default_params={"temperature": 0.7},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result["temperature"] == 0.7
+
+    def test_top_p(self):
+        result = self.config.map_openai_params(
+            non_default_params={"top_p": 0.5},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result["top_p"] == 0.5
+
+    def test_max_tokens(self):
+        result = self.config.map_openai_params(
+            non_default_params={"max_tokens": 100},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result["max_tokens"] == 100
+
+    def test_max_completion_tokens(self):
+        result = self.config.map_openai_params(
+            non_default_params={"max_completion_tokens": 200},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result["max_tokens"] == 200
+
+    def test_stop_is_dropped(self):
+        result = self.config.map_openai_params(
+            non_default_params={"stop": ["\n\n"]},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert "stop" not in result
+
+    def test_tools_converted_to_functions(self):
+        tools = [
+            {
+                "type": "function",
+                "function": {
+                    "name": "get_weather",
+                    "description": "Get weather",
+                    "parameters": {"type": "object"},
+                },
+            }
+        ]
+        result = self.config.map_openai_params(
+            non_default_params={"tools": tools},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert "functions" in result
+        assert result["functions"] == [
+            {"name": "get_weather", "description": "Get weather", "parameters": {"type": "object"}}
+        ]
+
+    def test_tool_choice_auto(self):
+        result = self.config.map_openai_params(
+            non_default_params={"tool_choice": "auto"},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result.get("function_call") == "auto"
+
+    def test_tool_choice_none(self):
+        result = self.config.map_openai_params(
+            non_default_params={"tool_choice": "none"},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result.get("function_call") == "none"
+
+    def test_tool_choice_required(self):
+        result = self.config.map_openai_params(
+            non_default_params={"tool_choice": "required"},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result.get("function_call") == "auto"
+
+    def test_tool_choice_dict(self):
+        result = self.config.map_openai_params(
+            non_default_params={
+                "tool_choice": {
+                    "type": "function",
+                    "function": {"name": "get_weather"},
+                }
+            },
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result.get("function_call") == {"name": "get_weather"}
+
+    def test_functions(self):
+        funcs = [{"name": "my_func", "description": "desc", "parameters": {}}]
+        result = self.config.map_openai_params(
+            non_default_params={"functions": funcs},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result["functions"] == funcs
+
+    def test_function_call(self):
+        result = self.config.map_openai_params(
+            non_default_params={"function_call": {"name": "my_func"}},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result["function_call"] == {"name": "my_func"}
+
+    def test_response_format_json_schema(self):
+        response_format = {
+            "type": "json_schema",
+            "json_schema": {
+                "name": "test_schema",
+                "schema": {"type": "object", "properties": {"name": {"type": "string"}}},
+            },
+        }
+        result = self.config.map_openai_params(
+            non_default_params={"response_format": response_format},
+            optional_params={"functions": []},
+            model="GigaChat",
+            drop_params=False,
+        )
+        # Should add a function for the schema
+        assert len(result["functions"]) == 1
+        assert result["functions"][0]["name"] == "test_schema"
+        assert result["function_call"] == {"name": "test_schema"}
+        assert result["_structured_output"] is True
+
+
+class TestConvertToolsToFunctions:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    def test_converts_function_tools_only(self):
+        tools = [
+            {"type": "function", "function": {"name": "a", "description": "d", "parameters": {}}},
+            {"type": "code_interpreter"},  # should be ignored
+        ]
+        result = self.config._convert_tools_to_functions(tools)
+        assert len(result) == 1
+        assert result[0]["name"] == "a"
+
+    def test_empty_tools(self):
+        assert self.config._convert_tools_to_functions([]) == []
+
+
+class TestMapToolChoice:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    def test_none(self):
+        assert self.config._map_tool_choice("none") == "none"
+
+    def test_auto(self):
+        assert self.config._map_tool_choice("auto") == "auto"
+
+    def test_required(self):
+        assert self.config._map_tool_choice("required") == "auto"
+
+    def test_dict_with_function(self):
+        result = self.config._map_tool_choice(
+            {"type": "function", "function": {"name": "get_weather"}}
+        )
+        assert result == {"name": "get_weather"}
+
+    def test_dict_without_name(self):
+        result = self.config._map_tool_choice(
+            {"type": "function", "function": {}}
+        )
+        assert result is None
+
+    def test_unknown_value(self):
+        assert self.config._map_tool_choice("unknown") is None
+
+
+class TestTransformMessages:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    def test_developer_role_to_system(self):
+        result = self.config._transform_messages(
+            [{"role": "developer", "content": "be helpful"}]
+        )
+        assert result[0]["role"] == "system"
+        assert result[0]["content"] == "be helpful"
+
+    def test_system_message_not_first_becomes_user(self):
+        result = self.config._transform_messages([
+            {"role": "user", "content": "hi"},
+            {"role": "system", "content": "instruction"},
+        ])
+        assert result[0]["role"] == "user"
+        assert result[1]["role"] == "user"
+        assert result[1]["content"] == "instruction"
+
+    def test_tool_role_to_function(self):
+        result = self.config._transform_messages([
+            {"role": "tool", "content": '{"result": "ok"}'}
+        ])
+        assert result[0]["role"] == "function"
+
+    def test_tool_role_content_wraps_non_json(self):
+        result = self.config._transform_messages([
+            {"role": "tool", "content": "plain text"}
+        ])
+        assert result[0]["role"] == "function"
+        assert is_valid_json(result[0]["content"])
+
+    def test_none_content_becomes_empty_string(self):
+        result = self.config._transform_messages([
+            {"role": "user", "content": None}
+        ])
+        assert result[0]["content"] == ""
+
+    def test_name_field_removed(self):
+        result = self.config._transform_messages([
+            {"role": "user", "content": "hi", "name": "John"}
+        ])
+        assert "name" not in result[0]
+
+    def test_tool_calls_converted_to_function_call(self):
+        result = self.config._transform_messages([
+            {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [
+                    {
+                        "id": "call_abc",
+                        "type": "function",
+                        "function": {
+                            "name": "get_weather",
+                            "arguments": '{"city": "London"}',
+                        },
+                    }
+                ],
+            }
+        ])
+        assert "tool_calls" not in result[0]
+        assert result[0]["function_call"]["name"] == "get_weather"
+        assert result[0]["function_call"]["arguments"] == {"city": "London"}
+
+    def test_tool_calls_with_dict_arguments(self):
+        result = self.config._transform_messages([
+            {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [
+                    {
+                        "id": "call_xyz",
+                        "type": "function",
+                        "function": {
+                            "name": "search",
+                            "arguments": {"query": "test"},
+                        },
+                    }
+                ],
+            }
+        ])
+        assert result[0]["function_call"]["arguments"] == {"query": "test"}
+
+    def test_list_content_multimodal(self):
+        content = [
+            {"type": "text", "text": "describe this"},
+            {
+                "type": "image_url",
+                "image_url": {"url": "https://example.com/img.jpg"},
+            },
+        ]
+        with patch.object(self.config, "_upload_image", return_value="file-123"):
+            result = self.config._transform_messages([
+                {"role": "user", "content": content}
+            ])
+        assert result[0]["content"] == "describe this"
+        assert result[0]["attachments"] == ["file-123"]
+
+    def test_list_content_with_image_url_string(self):
+        content = [
+            {"type": "text", "text": "look"},
+            {"type": "image_url", "image_url": "https://example.com/img.jpg"},
+        ]
+        with patch.object(self.config, "_upload_image", return_value="file-456"):
+            result = self.config._transform_messages([
+                {"role": "user", "content": content}
+            ])
+        assert result[0]["content"] == "look"
+        assert "file-456" in result[0]["attachments"]
+
+
+class TestTransformRequest:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    def test_builds_basic_request(self):
+        body = self.config.transform_request(
+            model="gigachat/GigaChat",
+            messages=[{"role": "user", "content": "hi"}],
+            optional_params={},
+            litellm_params={},
+            headers={},
+        )
+        assert body["model"] == "GigaChat"
+        assert len(body["messages"]) == 1
+        assert body["messages"][0]["content"] == "hi"
+
+    def test_model_prefix_stripped(self):
+        body = self.config.transform_request(
+            model="gigachat/GigaChat-Pro",
+            messages=[{"role": "user", "content": "hello"}],
+            optional_params={},
+            litellm_params={},
+            headers={},
+        )
+        assert body["model"] == "GigaChat-Pro"
+
+    def test_includes_optional_params(self):
+        body = self.config.transform_request(
+            model="gigachat/GigaChat",
+            messages=[{"role": "user", "content": "hi"}],
+            optional_params={
+                "temperature": 0.5,
+                "max_tokens": 100,
+                "stream": True,
+            },
+            litellm_params={},
+            headers={},
+        )
+        assert body["temperature"] == 0.5
+        assert body["max_tokens"] == 100
+        assert body["stream"] is True
+
+    def test_includes_functions(self):
+        body = self.config.transform_request(
+            model="gigachat/GigaChat",
+            messages=[{"role": "user", "content": "hi"}],
+            optional_params={
+                "functions": [{"name": "my_func"}],
+                "function_call": {"name": "my_func"},
+            },
+            litellm_params={},
+            headers={},
+        )
+        assert body["functions"] == [{"name": "my_func"}]
+        assert body["function_call"] == {"name": "my_func"}
+
+    def test_skips_unsupported_params(self):
+        body = self.config.transform_request(
+            model="gigachat/GigaChat",
+            messages=[{"role": "user", "content": "hi"}],
+            optional_params={"n": 2, "user": "abc"},
+            litellm_params={},
+            headers={},
+        )
+        assert "n" not in body
+        assert "user" not in body
+
+
+class TestTransformResponse:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    def test_basic_response(self):
+        raw = _make_httpx_response({
+            "id": "chatcmpl-123",
+            "created": 1700000000,
+            "model": "GigaChat",
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {"role": "assistant", "content": "Hello!"},
+                    "finish_reason": "stop",
+                }
+            ],
+            "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8},
+        })
+        model_response = ModelResponse()
+        result = self.config.transform_response(
+            model="gigachat/GigaChat",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=MagicMock(),
+            request_data={},
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            encoding=None,
+        )
+        assert result.choices[0].message.content == "Hello!"
+        assert result.choices[0].finish_reason == "stop"
+        assert result.usage.prompt_tokens == 5
+        assert result.usage.total_tokens == 8
+
+    def test_function_call_into_tool_calls(self):
+        raw = _make_httpx_response({
+            "id": "chatcmpl-456",
+            "created": 1700000000,
+            "model": "GigaChat",
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "content": None,
+                        "function_call": {
+                            "name": "get_weather",
+                            "arguments": {"city": "Moscow"},
+                        },
+                    },
+                    "finish_reason": "function_call",
+                }
+            ],
+            "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+        })
+        model_response = ModelResponse()
+        result = self.config.transform_response(
+            model="gigachat/GigaChat",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=MagicMock(),
+            request_data={},
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            encoding=None,
+        )
+        assert result.choices[0].finish_reason == "tool_calls"
+        tool_calls = result.choices[0].message.tool_calls
+        assert tool_calls is not None
+        assert len(tool_calls) == 1
+        assert tool_calls[0].function.name == "get_weather"
+        assert '{"city": "Moscow"}' in tool_calls[0].function.arguments
+
+    def test_function_call_structured_output(self):
+        raw = _make_httpx_response({
+            "id": "chatcmpl-789",
+            "created": 1700000000,
+            "model": "GigaChat",
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "content": None,
+                        "function_call": {
+                            "name": "test_schema",
+                            "arguments": {"name": "John"},
+                        },
+                    },
+                    "finish_reason": "function_call",
+                }
+            ],
+            "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+        })
+        model_response = ModelResponse()
+        result = self.config.transform_response(
+            model="gigachat/GigaChat",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=MagicMock(),
+            request_data={},
+            messages=[],
+            optional_params={"_structured_output": True},
+            litellm_params={},
+            encoding=None,
+        )
+        # Structured output: function_call -> content
+        assert result.choices[0].finish_reason == "stop"
+        assert result.choices[0].message.content is not None
+        assert '"name": "John"' in result.choices[0].message.content
+
+    def test_function_call_string_arguments(self):
+        raw = _make_httpx_response({
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "function_call": {
+                            "name": "get_weather",
+                            "arguments": '{"city": "Moscow"}',
+                        },
+                    },
+                    "finish_reason": "function_call",
+                }
+            ],
+            "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+        })
+        model_response = ModelResponse()
+        result = self.config.transform_response(
+            model="gigachat/GigaChat",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=MagicMock(),
+            request_data={},
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            encoding=None,
+        )
+        tc = result.choices[0].message.tool_calls[0]
+        assert '{"city": "Moscow"}' in tc.function.arguments
+
+    def test_cleans_up_gigachat_specific_fields(self):
+        raw = _make_httpx_response({
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "content": "done",
+                        "functions_state_id": "some-state",
+                    },
+                    "finish_reason": "stop",
+                }
+            ],
+            "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+        })
+        model_response = ModelResponse()
+        result = self.config.transform_response(
+            model="gigachat/GigaChat",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=MagicMock(),
+            request_data={},
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            encoding=None,
+        )
+        # functions_state_id should have been removed from the message data
+        assert result.choices[0].message.content == "done"
+
+    def test_raises_on_invalid_json(self):
+        raw = httpx.Response(
+            status_code=500,
+            headers={"content-type": "text/plain"},
+            content=b"not json",
+            request=httpx.Request("POST", "https://example.com"),
+        )
+        model_response = ModelResponse()
+        with pytest.raises(GigaChatError) as exc_info:
+            self.config.transform_response(
+                model="gigachat/GigaChat",
+                raw_response=raw,
+                model_response=model_response,
+                logging_obj=MagicMock(),
+                request_data={},
+                messages=[],
+                optional_params={},
+                litellm_params={},
+                encoding=None,
+            )
+        assert "Invalid JSON response" in str(exc_info.value.message)
+
+    def test_empty_choices(self):
+        raw = _make_httpx_response({
+            "choices": [],
+            "usage": {},
+        })
+        model_response = ModelResponse()
+        result = self.config.transform_response(
+            model="gigachat/GigaChat",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=MagicMock(),
+            request_data={},
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            encoding=None,
+        )
+        assert result.choices == []
+
+    def test_function_call_with_non_dict_arguments(self):
+        raw = _make_httpx_response({
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "function_call": {
+                            "name": "say_hello",
+                            "arguments": "hello",
+                        },
+                    },
+                    "finish_reason": "function_call",
+                }
+            ],
+            "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+        })
+        model_response = ModelResponse()
+        result = self.config.transform_response(
+            model="gigachat/GigaChat",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=MagicMock(),
+            request_data={},
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            encoding=None,
+        )
+        tc = result.choices[0].message.tool_calls[0]
+        assert tc.function.arguments == "hello"
+
+
+class TestGetModelResponseIterator:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    def test_returns_gigachat_iterator_sync(self):
+        from litellm.llms.gigachat.chat.streaming import (
+            GigaChatModelResponseIterator,
+        )
+
+        result = self.config.get_model_response_iterator(
+            streaming_response=iter(["data"]),
+            sync_stream=True,
+            json_mode=False,
+        )
+        assert isinstance(result, GigaChatModelResponseIterator)
+
+
+class TestGetErrorClass:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    def test_returns_gigachat_error(self):
+        error = self.config.get_error_class(
+            error_message="something went wrong",
+            status_code=400,
+            headers={"x-request-id": "abc"},
+        )
+        assert isinstance(error, GigaChatError)
+        assert error.status_code == 400
+        assert error.message == "something went wrong"
+        assert error.headers == {"x-request-id": "abc"}
+
+
+class TestUploadImage:
+    def setup_method(self):
+        self.config = GigaChatConfig()
+
+    @patch(f"{TRANSFORM_MODULE}.upload_file_sync", return_value="file-uploaded")
+    def test_upload_image_success(self, mock_upload):
+        self.config._current_credentials = "creds"
+        self.config._current_api_base = "https://api.example.com"
+        result = self.config._upload_image("https://example.com/img.jpg")
+        assert result == "file-uploaded"
+        mock_upload.assert_called_once_with(
+            image_url="https://example.com/img.jpg",
+            credentials="creds",
+            api_base="https://api.example.com",
+        )
+
+    @patch(f"{TRANSFORM_MODULE}.upload_file_sync", side_effect=Exception("fail"))
+    def test_upload_image_failure_returns_none(self, mock_upload):
+        result = self.config._upload_image("https://example.com/img.jpg")
+        assert result is None

tests/test_litellm/llms/gigachat/embedding/test_gigachat_embedding_transformation.py

@@ -0,0 +1,375 @@
+"""
+Unit tests for GigaChat embedding transformation.
+
+Tests GigaChatEmbeddingConfig covering get_config, get_supported_openai_params,
+map_openai_params, _get_openai_compatible_provider_info, get_complete_url,
+transform_embedding_request, transform_embedding_response, validate_environment,
+and get_error_class.
+"""
+
+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 import LlmProviders
+from litellm.llms.gigachat.embedding.transformation import (
+    GigaChatEmbeddingConfig,
+    GigaChatEmbeddingError,
+)
+from litellm.types.utils import EmbeddingResponse
+
+TRANSFORM_MODULE = "litellm.llms.gigachat.embedding.transformation"
+
+
+def _make_httpx_response(body: dict, status_code: int = 200) -> httpx.Response:
+    return httpx.Response(
+        status_code=status_code,
+        headers={"content-type": "application/json"},
+        content=json.dumps(body).encode("utf-8"),
+        request=httpx.Request("POST", "https://gigachat.devices.sberbank.ru/api/v1/embeddings"),
+    )
+
+
+# ---------------------------------------------------------------------------
+# GigaChatEmbeddingConfig
+# ---------------------------------------------------------------------------
+
+
+class TestGetConfig:
+    def setup_method(self):
+        self.config = GigaChatEmbeddingConfig()
+
+    def test_contains_only_abc_impl(self):
+        """get_config returns ABC internal data due to inheritance."""
+        result = self.config.get_config()
+        # The only key should be _abc_impl from ABC base class
+        assert set(result.keys()) == {"_abc_impl"}
+
+
+class TestGetSupportedOpenAiParams:
+    def setup_method(self):
+        self.config = GigaChatEmbeddingConfig()
+
+    def test_returns_empty_list(self):
+        params = self.config.get_supported_openai_params("GigaChat")
+        assert params == []
+
+
+class TestMapOpenAiParams:
+    def setup_method(self):
+        self.config = GigaChatEmbeddingConfig()
+
+    def test_returns_optional_params_unchanged(self):
+        result = self.config.map_openai_params(
+            non_default_params={"model": "test"},
+            optional_params={"temperature": 0.5},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result == {"temperature": 0.5}
+
+    def test_returns_empty_dict_when_no_optional_params(self):
+        result = self.config.map_openai_params(
+            non_default_params={},
+            optional_params={},
+            model="GigaChat",
+            drop_params=False,
+        )
+        assert result == {}
+
+
+class TestGetOpenaiCompatibleProviderInfo:
+    def setup_method(self):
+        self.config = GigaChatEmbeddingConfig()
+
+    def test_returns_gigachat_provider(self):
+        provider, api_base, api_key = self.config._get_openai_compatible_provider_info(
+            api_base="https://api.example.com", api_key="test-key"
+        )
+        assert provider == LlmProviders.GIGACHAT.value
+        assert api_base == "https://api.example.com"
+        assert api_key == "test-key"
+
+    def test_resolves_api_base_when_none(self):
+        provider, api_base, api_key = self.config._get_openai_compatible_provider_info(
+            api_base=None, api_key="key"
+        )
+        assert api_base is not None
+        assert api_base.endswith("/api/v1")
+
+    def test_returns_none_api_key(self):
+        _, _, api_key = self.config._get_openai_compatible_provider_info(
+            api_base="https://example.com", api_key=None
+        )
+        assert api_key is None
+
+
+class TestGetCompleteUrl:
+    def setup_method(self):
+        self.config = GigaChatEmbeddingConfig()
+
+    def test_default_url(self):
+        url = self.config.get_complete_url(
+            api_base=None, api_key=None, model="GigaChat",
+            optional_params={}, litellm_params={},
+        )
+        assert url.endswith("/embeddings")
+
+    def test_custom_api_base(self):
+        url = self.config.get_complete_url(
+            api_base="https://custom.example.com", api_key=None, model="GigaChat",
+            optional_params={}, litellm_params={},
+        )
+        assert url == "https://custom.example.com/embeddings"
+
+    def test_trailing_slash_api_base(self):
+        url = self.config.get_complete_url(
+            api_base="https://custom.example.com/", api_key=None, model="GigaChat",
+            optional_params={}, litellm_params={},
+        )
+        # get_api_base doesn't strip slash, so we get double slash
+        assert url == "https://custom.example.com//embeddings"
+
+
+class TestTransformEmbeddingRequest:
+    def setup_method(self):
+        self.config = GigaChatEmbeddingConfig()
+
+    def test_string_input(self):
+        result = self.config.transform_embedding_request(
+            model="gigachat/Embeddings",
+            input="hello world",
+            optional_params={},
+            headers={},
+        )
+        assert result == {"model": "Embeddings", "input": ["hello world"]}
+
+    def test_list_input(self):
+        result = self.config.transform_embedding_request(
+            model="gigachat/Embeddings",
+            input=["text1", "text2"],
+            optional_params={},
+            headers={},
+        )
+        assert result == {"model": "Embeddings", "input": ["text1", "text2"]}
+
+    def test_strips_gigachat_prefix(self):
+        result = self.config.transform_embedding_request(
+            model="gigachat/GigaChat-Pro",
+            input="test",
+            optional_params={},
+            headers={},
+        )
+        assert result["model"] == "GigaChat-Pro"
+
+    def test_model_without_prefix(self):
+        result = self.config.transform_embedding_request(
+            model="Embeddings",
+            input="test",
+            optional_params={},
+            headers={},
+        )
+        assert result["model"] == "Embeddings"
+
+
+class TestTransformEmbeddingResponse:
+    def setup_method(self):
+        self.config = GigaChatEmbeddingConfig()
+        self.logging_obj = MagicMock()
+
+    def _make_gigachat_response(self, data: list[dict]) -> httpx.Response:
+        return _make_httpx_response({
+            "object": "list",
+            "data": data,
+            "model": "Embeddings",
+        })
+
+    def test_basic_response(self):
+        raw = self._make_gigachat_response([
+            {
+                "object": "embedding",
+                "embedding": [0.1, 0.2, 0.3],
+                "index": 0,
+            }
+        ])
+        model_response = EmbeddingResponse()
+        result = self.config.transform_embedding_response(
+            model="gigachat/Embeddings",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=self.logging_obj,
+            api_key="test-key",
+            request_data={"input": ["text"]},
+            optional_params={},
+            litellm_params={},
+        )
+        assert result.object == "list"
+        assert len(result.data) == 1
+        assert result.data[0]["embedding"] == [0.1, 0.2, 0.3]
+        assert result.data[0]["index"] == 0
+        assert result.usage.prompt_tokens == 0
+        assert result.usage.total_tokens == 0
+
+    def test_aggregates_per_embedding_usage(self):
+        raw = self._make_gigachat_response([
+            {
+                "object": "embedding",
+                "embedding": [0.1, 0.2],
+                "index": 0,
+                "usage": {"prompt_tokens": 5},
+            },
+            {
+                "object": "embedding",
+                "embedding": [0.3, 0.4],
+                "index": 1,
+                "usage": {"prompt_tokens": 7},
+            },
+        ])
+        model_response = EmbeddingResponse()
+        result = self.config.transform_embedding_response(
+            model="gigachat/Embeddings",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=self.logging_obj,
+            api_key="test-key",
+            request_data={"input": ["a", "b"]},
+            optional_params={},
+            litellm_params={},
+        )
+        # Total should be sum of per-embedding prompt_tokens
+        assert result.usage.prompt_tokens == 12
+        assert result.usage.total_tokens == 12
+        # Usage should be removed from individual embedding data
+        assert "usage" not in result.data[0]
+        assert "usage" not in result.data[1]
+
+    def test_usage_removed_from_individual_embeddings(self):
+        raw = self._make_gigachat_response([
+            {
+                "object": "embedding",
+                "embedding": [0.5],
+                "index": 0,
+                "usage": {"prompt_tokens": 3},
+            }
+        ])
+        model_response = EmbeddingResponse()
+        result = self.config.transform_embedding_response(
+            model="gigachat/Embeddings",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=self.logging_obj,
+            api_key="key",
+            request_data={"input": ["x"]},
+            optional_params={},
+            litellm_params={},
+        )
+        # usage should NOT be in the final EmbeddingResponse data items
+        for emb in result.data:
+            assert "usage" not in emb
+
+    def test_passes_model_from_response(self):
+        raw = self._make_gigachat_response([
+            {"object": "embedding", "embedding": [0.1], "index": 0},
+        ])
+        model_response = EmbeddingResponse()
+        result = self.config.transform_embedding_response(
+            model="gigachat/Embeddings",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=self.logging_obj,
+            api_key="key",
+            request_data={"input": ["x"]},
+            optional_params={},
+            litellm_params={},
+        )
+        assert result.model == "Embeddings"
+
+    def test_calls_logging_post_call(self):
+        raw = self._make_gigachat_response([
+            {"object": "embedding", "embedding": [0.1], "index": 0},
+        ])
+        model_response = EmbeddingResponse()
+        self.config.transform_embedding_response(
+            model="gigachat/Embeddings",
+            raw_response=raw,
+            model_response=model_response,
+            logging_obj=self.logging_obj,
+            api_key="test-api-key",
+            request_data={"input": ["hello"]},
+            optional_params={},
+            litellm_params={},
+        )
+        self.logging_obj.post_call.assert_called_once()
+        args = self.logging_obj.post_call.call_args.kwargs
+        assert args["api_key"] == "test-api-key"
+        assert args["input"] == ["hello"]
+
+
+class TestValidateEnvironment:
+    def setup_method(self):
+        self.config = GigaChatEmbeddingConfig()
+
+    @patch(f"{TRANSFORM_MODULE}.get_access_token", return_value="test-token")
+    def test_sets_oauth_headers(self, mock_get_token):
+        headers = self.config.validate_environment(
+            headers={},
+            model="GigaChat",
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            api_key="creds",
+            api_base="https://api.example.com",
+        )
+        assert headers["Authorization"] == "Bearer test-token"
+        assert headers["Content-Type"] == "application/json"
+        mock_get_token.assert_called_once_with(credentials="creds", litellm_params={})
+
+    @patch(f"{TRANSFORM_MODULE}.get_access_token", return_value="token")
+    def test_merges_custom_headers(self, mock_get_token):
+        headers = self.config.validate_environment(
+            headers={"X-Custom": "value"},
+            model="GigaChat",
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            api_key="creds",
+            api_base="https://api.example.com",
+        )
+        assert headers["Authorization"] == "Bearer token"
+        assert headers["Content-Type"] == "application/json"
+        assert headers["X-Custom"] == "value"
+
+    @patch(f"{TRANSFORM_MODULE}.get_access_token", return_value="token")
+    def test_custom_header_overwrites_default(self, mock_get_token):
+        headers = self.config.validate_environment(
+            headers={"Authorization": "Bearer custom"},
+            model="GigaChat",
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            api_key="creds",
+            api_base="https://api.example.com",
+        )
+        # Merge: default headers first, then custom headers on top
+        assert headers["Authorization"] == "Bearer custom"
+
+
+class TestGetErrorClass:
+    def setup_method(self):
+        self.config = GigaChatEmbeddingConfig()
+
+    def test_returns_gigachat_embedding_error(self):
+        error = self.config.get_error_class(
+            error_message="embedding failed",
+            status_code=400,
+            headers={"x-request-id": "abc"},
+        )
+        assert isinstance(error, GigaChatEmbeddingError)
+        assert error.status_code == 400
+        assert error.message == "embedding failed"

tests/test_litellm/llms/gigachat/passthrough/test_gigachat_passthrough_transformation.py

@@ -479,3 +479,131 @@ class TestGigaChatPassthroughConfig:
         config = GigaChatPassthroughConfig()
         result = config.get_models()
         assert result == []
+
+    def test_logging_non_streaming_chat_raises_when_no_config(self):
+        """Test raise when ProviderConfigManager returns None for chat."""
+        config = GigaChatPassthroughConfig()
+        logging_obj = MagicMock()
+
+        with patch(
+            "litellm.utils.ProviderConfigManager.get_provider_chat_config",
+            return_value=None,
+        ):
+            with pytest.raises(ValueError, match="No provider config found for model"):
+                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",
+                )
+
+    def test_logging_non_streaming_embedding_raises_when_no_config(self):
+        """Test raise when ProviderConfigManager returns None for embeddings."""
+        config = GigaChatPassthroughConfig()
+        logging_obj = MagicMock()
+
+        with patch(
+            "litellm.utils.ProviderConfigManager.get_provider_embedding_config",
+            return_value=None,
+        ):
+            with pytest.raises(ValueError, match="No provider config found for model"):
+                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",
+                )
+
+    def test_handle_logging_collected_chunks_with_model_response_stream_chunk(self):
+        """Test that a chunk returning ModelResponseStream from chunk_parser is handled.
+
+        Requires patching GigaChatModelResponseIterator.chunk_parser to return
+        a ModelResponseStream so the elif branch is exercised.
+        """
+        config = GigaChatPassthroughConfig()
+        logging_obj = MagicMock()
+
+        from litellm.types.utils import ModelResponseStream
+
+        stream_chunk = ModelResponseStream(
+            choices=[
+                {
+                    "index": 0,
+                    "delta": {"content": "streamed"},
+                    "finish_reason": None,
+                }
+            ]
+        )
+
+        chunks = [
+            '{"choices": [{"delta": {"content": "streamed"}, "index": 0}]}',
+            '{"choices": [{"delta": {}, "finish_reason": "stop", "index": 0}], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}}',
+        ]
+
+        with patch(
+            "litellm.llms.gigachat.passthrough.transformation.GigaChatModelResponseIterator.chunk_parser",
+            return_value=stream_chunk,
+        ):
+            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 == "streamedstreamed"
+
+    def test_handle_logging_collected_chunks_skips_unknown_chunk_type(self):
+        """Test that chunk_parser returning an unknown type is skipped."""
+        config = GigaChatPassthroughConfig()
+        logging_obj = MagicMock()
+
+        chunks = [
+            '{"choices": [{"delta": {"content": "good"}, "index": 0}]}',
+            '{"choices": [{"delta": {}, "finish_reason": "stop", "index": 0}], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}}',
+        ]
+
+        with patch(
+            "litellm.llms.gigachat.passthrough.transformation.GigaChatModelResponseIterator.chunk_parser",
+            return_value=12345,  # not dict and not ModelResponseStream
+        ):
+            result = config.handle_logging_collected_chunks(
+                all_chunks=chunks,
+                litellm_logging_obj=logging_obj,
+                model="gigachat/GigaChat",
+                custom_llm_provider="gigachat",
+                endpoint="chat/completions",
+            )
+
+        # All chunks skipped, returns None
+        assert result is None
+
+    def test_handle_logging_collected_chunks_skips_unsupported_chunk_type(self):
+        """Test that unsupported chunk types (int, float, etc.) are skipped."""
+        config = GigaChatPassthroughConfig()
+        logging_obj = MagicMock()
+
+        # The chunk is an int which doesn't match str/bytes/dict
+        chunks: list = [42, "not-a-real-chunk"]
+
+        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 result is None

tests/test_litellm/llms/gigachat/test_file_handler.py

@@ -0,0 +1,508 @@
+"""
+Unit tests for GigaChat file handler.
+
+Tests _get_url_hash, _parse_data_url, _download_image_sync, _download_image_async,
+upload_file_sync, and upload_file_async covering caching, base64 data URL decoding,
+network errors, and the full upload flow.
+"""
+
+import json
+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 import file_handler
+from litellm.llms.gigachat.file_handler import (
+    _file_cache,
+    _get_url_hash,
+    _parse_data_url,
+    upload_file_async,
+    upload_file_sync,
+)
+
+FILE_MODULE = "litellm.llms.gigachat.file_handler"
+
+# A valid 1x1 red PNG as base64
+_RED_PNG_B64 = (
+    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAA"
+    "DUlEQVQI12NgYPgPAAEDAQAR3X3ZAAAASUVORK5CYII="
+)
+_RED_PNG_DATA_URL = f"data:image/png;base64,{_RED_PNG_B64}"
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def _isolate_file_cache():
+    """Each test gets a fresh module-level file cache to avoid cross-test leakage."""
+    _file_cache.clear()
+    yield
+    _file_cache.clear()
+
+
+# ---------------------------------------------------------------------------
+# _get_url_hash
+# ---------------------------------------------------------------------------
+
+
+class TestGetUrlHash:
+    def test_returns_hex_string(self):
+        h = _get_url_hash("https://example.com/image.png")
+        assert isinstance(h, str)
+        assert len(h) == 64  # SHA-256
+
+    def test_different_urls_different_hashes(self):
+        h1 = _get_url_hash("https://example.com/a.png")
+        h2 = _get_url_hash("https://example.com/b.png")
+        assert h1 != h2
+
+    def test_same_url_same_hash(self):
+        h1 = _get_url_hash("https://example.com/image.png")
+        h2 = _get_url_hash("https://example.com/image.png")
+        assert h1 == h2
+
+
+# ---------------------------------------------------------------------------
+# _parse_data_url
+# ---------------------------------------------------------------------------
+
+
+class TestParseDataUrl:
+    def test_valid_base64_png(self):
+        result = _parse_data_url(_RED_PNG_DATA_URL)
+        assert result is not None
+        content_bytes, content_type, ext = result
+        assert content_type == "image/png"
+        assert ext == "png"
+        assert len(content_bytes) > 0
+
+    def test_valid_base64_jpeg(self):
+        # Simple valid base64 (24 chars, properly padded, no + or / chars)
+        valid_b64 = "aGVsbG8gd29ybGQhISEhIQ=="
+        data_url = f"data:image/jpeg;base64,{valid_b64}"
+        result = _parse_data_url(data_url)
+        assert result is not None
+        _, content_type, ext = result
+        assert content_type == "image/jpeg"
+        assert ext == "jpeg"
+
+    def test_valid_base64_with_semicolon_in_type(self):
+        """Data URLs with charset before base64 segment do not match the regex."""
+        # The regex `data:([^;]+);base64,(.+)` requires the pattern to be
+        # `data:<type>;base64,<data>`. If `;charset=utf-8` appears before
+        # `;base64,`, the regex sees `data:image/png` as group 1 but then
+        # looks for `;base64,` immediately after — which isn't there because
+        # `;charset=utf-8;base64,` has extra text before `;base64,`
+        data_url = "data:image/png;charset=utf-8;base64," + _RED_PNG_B64
+        result = _parse_data_url(data_url)
+        assert result is None
+
+    def test_invalid_data_url_returns_none(self):
+        assert _parse_data_url("not-a-data-url") is None
+
+    def test_empty_base64_returns_none(self):
+        """Empty base64 data (nothing after comma) does not match regex `(.+)`."""
+        assert _parse_data_url("data:image/png;base64,") is None
+
+    def test_missing_base64_segment(self):
+        assert _parse_data_url("data:image/png;base64") is None
+
+    def test_unknown_extension_falls_back_to_jpg(self):
+        data_url = "data:application/octet-stream;base64," + _RED_PNG_B64
+        result = _parse_data_url(data_url)
+        assert result is not None
+        _, content_type, ext = result
+        assert content_type == "application/octet-stream"
+        # The extension is derived from content_type.split("/")[-1].split(";")[0]
+        # which gives "octet-stream", not "jpg"
+        assert ext == "octet-stream"
+
+
+# ---------------------------------------------------------------------------
+# _download_image_sync
+# ---------------------------------------------------------------------------
+
+
+class TestDownloadImageSync:
+    @patch(f"{FILE_MODULE}._get_httpx_client")
+    def test_downloads_image_successfully(self, mock_get_client):
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.content = b"fake-image-bytes"
+        mock_response.headers = {"content-type": "image/jpeg"}
+        mock_client.get.return_value = mock_response
+        mock_get_client.return_value = mock_client
+
+        content_bytes, content_type, ext = file_handler._download_image_sync("https://example.com/img.jpg")
+
+        assert content_bytes == b"fake-image-bytes"
+        assert content_type == "image/jpeg"
+        assert ext == "jpeg"
+        mock_client.get.assert_called_once_with("https://example.com/img.jpg")
+
+    @patch(f"{FILE_MODULE}._get_httpx_client")
+    def test_raises_on_http_error(self, mock_get_client):
+        mock_client = MagicMock()
+        mock_client.get.side_effect = httpx.HTTPStatusError(
+            "Not Found",
+            request=httpx.Request("GET", "https://example.com/404"),
+            response=httpx.Response(status_code=404, request=httpx.Request("GET", "https://example.com/404")),
+        )
+        mock_get_client.return_value = mock_client
+
+        with pytest.raises(httpx.HTTPStatusError):
+            file_handler._download_image_sync("https://example.com/404")
+
+    @patch(f"{FILE_MODULE}._get_httpx_client")
+    def test_parse_content_type_fallback(self, mock_get_client):
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.content = b"data"
+        mock_response.headers = {}
+        mock_client.get.return_value = mock_response
+        mock_get_client.return_value = mock_client
+
+        _, content_type, ext = file_handler._download_image_sync("https://example.com/img")
+
+        assert content_type == "image/jpeg"
+        assert ext == "jpeg"
+
+    @patch(f"{FILE_MODULE}._get_httpx_client")
+    def test_extracts_extension_from_parametrized_type(self, mock_get_client):
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.content = b"data"
+        mock_response.headers = {"content-type": "image/png; charset=utf-8"}
+        mock_client.get.return_value = mock_response
+        mock_get_client.return_value = mock_client
+
+        _, _, ext = file_handler._download_image_sync("https://example.com/img.png")
+
+        assert ext == "png"
+
+
+# ---------------------------------------------------------------------------
+# _download_image_async
+# ---------------------------------------------------------------------------
+
+
+class TestDownloadImageAsync:
+    @pytest.mark.asyncio
+    @patch(f"{FILE_MODULE}.get_async_httpx_client")
+    async def test_downloads_image_successfully(self, mock_get_client):
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.content = b"fake-image-bytes"
+        mock_response.headers = {"content-type": "image/webp"}
+        mock_client.get = AsyncMock(return_value=mock_response)
+        mock_get_client.return_value = mock_client
+
+        content_bytes, content_type, ext = await file_handler._download_image_async(
+            "https://example.com/img.webp"
+        )
+
+        assert content_bytes == b"fake-image-bytes"
+        assert content_type == "image/webp"
+        assert ext == "webp"
+        mock_client.get.assert_called_once_with("https://example.com/img.webp")
+
+    @pytest.mark.asyncio
+    @patch(f"{FILE_MODULE}.get_async_httpx_client")
+    async def test_raises_on_http_error(self, mock_get_client):
+        mock_client = MagicMock()
+        mock_client.get = AsyncMock(
+            side_effect=httpx.HTTPStatusError(
+                "Forbidden",
+                request=httpx.Request("GET", "https://example.com/403"),
+                response=httpx.Response(status_code=403, request=httpx.Request("GET", "https://example.com/403")),
+            )
+        )
+        mock_get_client.return_value = mock_client
+
+        with pytest.raises(httpx.HTTPStatusError):
+            await file_handler._download_image_async("https://example.com/403")
+
+
+# ---------------------------------------------------------------------------
+# upload_file_sync
+# ---------------------------------------------------------------------------
+
+
+class TestUploadFileSync:
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    @patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
+    @patch(f"{FILE_MODULE}._get_httpx_client")
+    def test_uploads_base64_image_and_caches(
+        self, mock_get_client, mock_get_token, mock_get_api_base
+    ):
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"id": "file-12345"}
+        mock_response.raise_for_status = MagicMock()
+        mock_client.post.return_value = mock_response
+        mock_get_client.return_value = mock_client
+
+        result = upload_file_sync(
+            image_url=_RED_PNG_DATA_URL,
+            credentials="creds",
+            api_base="https://custom.example.com",
+        )
+
+        assert result == "file-12345"
+        # Verify it was cached
+        url_hash = _get_url_hash(_RED_PNG_DATA_URL)
+        assert _file_cache[url_hash] == "file-12345"
+
+        # Check the upload request — url is passed as first positional arg
+        call_args = mock_client.post.call_args
+        assert call_args.args[0] == "https://api.example.com/files"
+        assert call_args.kwargs["headers"]["Authorization"] == "Bearer test-token"
+        # Verify purpose
+        assert call_args.kwargs["data"] == {"purpose": "general"}
+        # Verify a file was attached
+        assert "file" in call_args.kwargs["files"]
+
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    @patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
+    @patch(f"{FILE_MODULE}._get_httpx_client")
+    def test_returns_cached_file_id(
+        self, mock_get_client, mock_get_token, mock_get_api_base
+    ):
+        # Pre-populate the cache
+        url_hash = _get_url_hash(_RED_PNG_DATA_URL)
+        _file_cache[url_hash] = "cached-file-id"
+
+        result = upload_file_sync(image_url=_RED_PNG_DATA_URL, credentials="creds")
+
+        assert result == "cached-file-id"
+        # No upload call was made
+        mock_get_client.return_value.post.assert_not_called()
+
+    @patch(f"{FILE_MODULE}._get_httpx_client")
+    @patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    @patch(f"{FILE_MODULE}._download_image_sync")
+    def test_downloads_and_uploads_url_image(
+        self, mock_download, mock_get_api_base, mock_get_token, mock_get_client
+    ):
+        mock_download.return_value = (b"remote-bytes", "image/png", "png")
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"id": "file-remote"}
+        mock_response.raise_for_status = MagicMock()
+        mock_client.post.return_value = mock_response
+        mock_get_client.return_value = mock_client
+
+        result = upload_file_sync(
+            image_url="https://example.com/remote.png", credentials="creds"
+        )
+
+        assert result == "file-remote"
+        mock_download.assert_called_once_with("https://example.com/remote.png")
+
+    @patch(f"{FILE_MODULE}._get_httpx_client")
+    @patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    def test_returns_none_on_upload_failure(
+        self, mock_get_api_base, mock_get_token, mock_get_client
+    ):
+        mock_client = MagicMock()
+        mock_client.post.side_effect = httpx.HTTPStatusError(
+            "Bad Request",
+            request=httpx.Request("POST", "https://api.example.com/files"),
+            response=httpx.Response(status_code=400, request=httpx.Request("POST", "https://api.example.com/files")),
+        )
+        mock_get_client.return_value = mock_client
+
+        # upload_file_sync catches all exceptions and returns None
+        result = upload_file_sync(
+            image_url=_RED_PNG_DATA_URL, credentials="creds"
+        )
+
+        assert result is None
+
+    @patch(f"{FILE_MODULE}._get_httpx_client")
+    @patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    def test_returns_none_when_response_missing_id(
+        self, mock_get_api_base, mock_get_token, mock_get_client
+    ):
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"status": "ok"}  # no "id" key
+        mock_response.raise_for_status = MagicMock()
+        mock_client.post.return_value = mock_response
+        mock_get_client.return_value = mock_client
+
+        result = upload_file_sync(
+            image_url=_RED_PNG_DATA_URL, credentials="creds"
+        )
+
+        assert result is None
+
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    @patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
+    @patch(f"{FILE_MODULE}._get_httpx_client")
+    def test_uploads_without_optional_args(
+        self, mock_get_client, mock_get_token, mock_get_api_base
+    ):
+        """Verify that credentials, api_base, and litellm_params are optional."""
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"id": "file-no-args"}
+        mock_response.raise_for_status = MagicMock()
+        mock_client.post.return_value = mock_response
+        mock_get_client.return_value = mock_client
+
+        result = upload_file_sync(image_url=_RED_PNG_DATA_URL)
+
+        assert result == "file-no-args"
+        # Should still have called get_access_token without args
+        mock_get_token.assert_called_once_with(credentials=None, litellm_params=None)
+
+
+# ---------------------------------------------------------------------------
+# upload_file_async
+# ---------------------------------------------------------------------------
+
+
+class TestUploadFileAsync:
+    @pytest.mark.asyncio
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    @patch(f"{FILE_MODULE}.get_access_token_async", return_value="test-token-async")
+    @patch(f"{FILE_MODULE}.get_async_httpx_client")
+    async def test_uploads_base64_image_and_caches(
+        self, mock_get_client, mock_get_token, mock_get_api_base
+    ):
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json = MagicMock(return_value={"id": "async-file-1"})
+        mock_response.raise_for_status = MagicMock()
+        mock_client.post = AsyncMock(return_value=mock_response)
+        mock_get_client.return_value = mock_client
+
+        result = await upload_file_async(
+            image_url=_RED_PNG_DATA_URL,
+            credentials="creds",
+            api_base="https://custom.example.com",
+        )
+
+        assert result == "async-file-1"
+        # Verify cache
+        url_hash = _get_url_hash(_RED_PNG_DATA_URL)
+        assert _file_cache[url_hash] == "async-file-1"
+
+        # Check upload request details — url is first positional arg
+        call_args = mock_client.post.call_args
+        assert call_args.args[0] == "https://api.example.com/files"
+        assert call_args.kwargs["headers"]["Authorization"] == "Bearer test-token-async"
+        assert "purpose" in str(call_args.kwargs["data"])
+        assert "file" in call_args.kwargs["files"]
+
+    @pytest.mark.asyncio
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    @patch(f"{FILE_MODULE}.get_access_token_async", return_value="test-token-async")
+    @patch(f"{FILE_MODULE}.get_async_httpx_client")
+    async def test_returns_cached_file_id(
+        self, mock_get_client, mock_get_token, mock_get_api_base
+    ):
+        url_hash = _get_url_hash(_RED_PNG_DATA_URL)
+        _file_cache[url_hash] = "cached-async-id"
+
+        result = await upload_file_async(image_url=_RED_PNG_DATA_URL, credentials="creds")
+
+        assert result == "cached-async-id"
+        mock_get_client.return_value.post.assert_not_called()
+
+    @pytest.mark.asyncio
+    @patch(f"{FILE_MODULE}.get_async_httpx_client")
+    @patch(f"{FILE_MODULE}.get_access_token_async", return_value="test-token-async")
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    @patch(f"{FILE_MODULE}._download_image_async")
+    async def test_downloads_and_uploads_url_image(
+        self, mock_download, mock_get_api_base, mock_get_token, mock_get_client
+    ):
+        mock_download.return_value = (b"remote-bytes-async", "image/png", "png")
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json = MagicMock(return_value={"id": "async-file-remote"})
+        mock_response.raise_for_status = MagicMock()
+        mock_client.post = AsyncMock(return_value=mock_response)
+        mock_get_client.return_value = mock_client
+
+        result = await upload_file_async(
+            image_url="https://example.com/remote.png", credentials="creds"
+        )
+
+        assert result == "async-file-remote"
+        mock_download.assert_called_once_with("https://example.com/remote.png")
+
+    @pytest.mark.asyncio
+    @patch(f"{FILE_MODULE}.get_async_httpx_client")
+    @patch(f"{FILE_MODULE}.get_access_token_async", return_value="test-token-async")
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    async def test_returns_none_on_upload_failure(
+        self, mock_get_api_base, mock_get_token, mock_get_client
+    ):
+        mock_client = MagicMock()
+        mock_client.post = AsyncMock(
+            side_effect=httpx.HTTPStatusError(
+                "Bad Request",
+                request=httpx.Request("POST", "https://api.example.com/files"),
+                response=httpx.Response(status_code=400, request=httpx.Request("POST", "https://api.example.com/files")),
+            )
+        )
+        mock_get_client.return_value = mock_client
+
+        result = await upload_file_async(
+            image_url=_RED_PNG_DATA_URL, credentials="creds"
+        )
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    @patch(f"{FILE_MODULE}.get_async_httpx_client")
+    @patch(f"{FILE_MODULE}.get_access_token_async", return_value="test-token-async")
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    async def test_returns_none_when_response_missing_id(
+        self, mock_get_api_base, mock_get_token, mock_get_client
+    ):
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json = MagicMock(return_value={"status": "ok"})
+        mock_response.raise_for_status = MagicMock()
+        mock_client.post = AsyncMock(return_value=mock_response)
+        mock_get_client.return_value = mock_client
+
+        result = await upload_file_async(
+            image_url=_RED_PNG_DATA_URL, credentials="creds"
+        )
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    @patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
+    @patch(f"{FILE_MODULE}.get_access_token_async", return_value="test-token-async")
+    @patch(f"{FILE_MODULE}.get_async_httpx_client")
+    async def test_uploads_without_optional_args(
+        self, mock_get_client, mock_get_token, mock_get_api_base
+    ):
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json = MagicMock(return_value={"id": "async-no-args"})
+        mock_response.raise_for_status = MagicMock()
+        mock_client.post = AsyncMock(return_value=mock_response)
+        mock_get_client.return_value = mock_client
+
+        result = await upload_file_async(image_url=_RED_PNG_DATA_URL)
+
+        assert result == "async-no-args"
+        mock_get_token.assert_called_once_with(credentials=None, litellm_params=None)