test(bedrock): preserve legacy imports and exercise owned HTTP transport

tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_transformation.py

@@ -1,11 +1,11 @@
 import json
-from typing import Final
+from typing import Final, Literal
 from unittest.mock import MagicMock, patch
 
 import httpx
 import pytest
-import respx
-from httpx import Headers
+from pydantic import JsonValue
+from typing_extensions import ReadOnly, TypedDict
 
 import litellm
 from litellm.constants import (
@@ -6355,7 +6355,20 @@ def test_json_mode_keeps_the_thinking_that_preceded_the_json_tool_call():
     assert [block["signature"] for block in message.thinking_blocks] == [thinking_block["signature"]]
 
 
-def _json_mode_message_body(stop_reason: str, stop_details: dict | None) -> dict:
+class JsonModeMessageBody(TypedDict):
+    id: ReadOnly[str]
+    type: ReadOnly[Literal["message"]]
+    role: ReadOnly[Literal["assistant"]]
+    model: ReadOnly[str]
+    stop_reason: ReadOnly[str]
+    stop_details: ReadOnly[dict[str, JsonValue] | None]
+    usage: ReadOnly[dict[str, int]]
+    content: ReadOnly[list[dict[str, JsonValue]]]
+
+
+def _json_mode_message_body(
+    stop_reason: str, stop_details: dict[str, JsonValue] | None
+) -> JsonModeMessageBody:
     return {
         "id": "msg_json_mode",
         "type": "message",
@@ -6370,9 +6383,9 @@ def _json_mode_message_body(stop_reason: str, stop_details: dict | None) -> dict
     }
 
 
-def _transform_json_mode_body(body: dict) -> litellm.ModelResponse:
+def _transform_json_mode_body(body: JsonModeMessageBody) -> litellm.ModelResponse:
     return AnthropicConfig().transform_parsed_response(
-        completion_response=body,
+        completion_response=dict(body),
         raw_response=httpx.Response(status_code=200, headers={}),
         model_response=litellm.ModelResponse(),
         json_mode=True,

tests/test_litellm/llms/bedrock/chat/test_bedrock_converse_handler.py

@@ -1,5 +1,3 @@
-import litellm
-
 """Tests for `BedrockConverseLLM.completion`.
 
 AWS credential resolution is stubbed so nothing reaches STS.
@@ -18,6 +16,7 @@ import pytest
 from botocore.credentials import Credentials
 from botocore.exceptions import ClientError
 
+import litellm
 from litellm.llms.bedrock.chat.converse_handler import BedrockConverseLLM
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
 from litellm.rust_bridge import configuration

tests/test_litellm/llms/bedrock/chat/test_converse_transformation.py

@@ -13,14 +13,12 @@ import litellm
 from litellm import ModelResponse, Router, completion
 from litellm.litellm_core_utils.prompt_templates.factory import (
     _bedrock_converse_messages_pt,
-    _bedrock_tools_pt,
 )
 from litellm.llms.anthropic.chat.transformation import (
     NATIVE_OUTPUT_MAX_GRAMMAR_BRANCHES,
     NATIVE_OUTPUT_MAX_OPTIONAL_PARAMETERS,
     NATIVE_OUTPUT_MAX_UNION_PARAMETERS,
 )
-from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
 from litellm.llms.bedrock.chat.converse_transformation import AmazonConverseConfig
 from litellm.llms.bedrock.common_utils import BEDROCK_NATIVE_OUTPUT_MAX_PROPERTIES
 from litellm.llms.custom_httpx.http_handler import HTTPHandler
@@ -4241,12 +4239,17 @@ _OPUS_4_8_INVOKE_RESPONSE: Final = {
 
 
 def _routed_request_body(router: Router, model_name: str, route: str) -> Mapping[str, object]:
-    response: Final = _mock_converse_response()
-    if route == "invoke":
-        response.json.return_value = _OPUS_4_8_INVOKE_RESPONSE
-        response.text = json.dumps(_OPUS_4_8_INVOKE_RESPONSE)
-    client: Final = HTTPHandler()
-    with patch.object(client, "post", return_value=response) as mock_post:
+    from queue import SimpleQueue
+
+    response_body: Final = _OPUS_4_8_INVOKE_RESPONSE if route == "invoke" else _mock_converse_response().json()
+    requests: Final = SimpleQueue[httpx.Request]()
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        requests.put(request)
+        return httpx.Response(200, json=response_body)
+
+    with httpx.Client(transport=httpx.MockTransport(respond)) as http_client:
+        client: Final = HTTPHandler(client=http_client)
         router.completion(
             model=model_name,
             messages=[{"role": "user", "content": "Name the capital of France."}],
@@ -4254,7 +4257,8 @@ def _routed_request_body(router: Router, model_name: str, route: str) -> Mapping
             thinking=ADAPTIVE_THINKING,
             client=client,
         )
-    return json.loads(mock_post.call_args.kwargs["data"])
+    assert requests.qsize() == 1
+    return json.loads(requests.get_nowait().content)
 
 
 def _sends_native_output(body: Mapping[str, object], route: str) -> bool:

tests/test_litellm/test_utils.py

@@ -5,7 +5,6 @@ import json
 import logging
 import os
 import queue
-import re
 import threading
 from collections.abc import Callable, Iterator
 from concurrent.futures import Future, ThreadPoolExecutor
@@ -33,7 +32,6 @@ from litellm.constants import DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
 from litellm.litellm_core_utils.thread_pool_executor import executor as logging_executor
-from litellm.llms.base_llm.audio_transcription.transformation import BaseAudioTranscriptionConfig
 from litellm.llms.base_llm.base_model_iterator import MockResponseIterator
 from litellm.proxy.utils import is_valid_api_key
 from litellm.types.integrations.custom_logger import HEADROOM_CONVERTED_STREAM_KEY
@@ -61,19 +59,12 @@ from litellm.utils import (
     _is_streaming_request,
     _run_success_deployment_hook_on_converted_chat_stream,
     _snapshot_exception_for_hook,
-    add_custom_logger_callback_to_specific_event,
     async_post_call_failure_deployment_hook,
     async_post_call_success_deployment_hook,
     calculate_max_parallel_requests,
-    check_valid_key,
     client,
-    filter_out_litellm_params,
-    get_applied_guardrails,
-    get_llm_provider,
     get_non_default_completion_params,
-    get_optional_params_embeddings,
     get_optional_params_image_gen,
-    get_potential_model_names,
     get_prompt_cache_min_tokens,
     is_cached_message,
     is_prompt_caching_valid_prompt,
@@ -586,6 +577,7 @@ def test_all_model_configs():
 
 
 def test_cohere_embedding_optional_params():
+    from litellm import get_optional_params_embeddings
 
     optional_params = get_optional_params_embeddings(
         model="embed-v4.0",
@@ -1607,6 +1599,7 @@ class TestProxyFunctionCalling:
         # For now, we expect False (current behavior), but document the limitation
         assert proxy_result is False, f"Current limitation: {proxy_model_with_hints} returns False without inference"
 
+
     def test_litellm_utils_supports_function_calling_import(self):
         """Test that supports_function_calling can be imported from litellm.utils."""
         try:
@@ -1626,6 +1619,7 @@ class TestProxyFunctionCalling:
         except Exception as e:
             pytest.fail(f"Failed to access litellm.supports_function_calling: {e}")
 
+
     def test_edge_cases_and_malformed_proxy_models(self):
         """Test edge cases and malformed proxy model names."""
         test_cases = [