test(bedrock): exercise owned HTTP transport and type copied fixtures

This commit is contained in:
Joshua Valluru 2026-10-20 19:32:39 -07:00
parent 5fff182577
commit 8a3f4ca463
3 changed files with 39 additions and 77 deletions

View file

@ -1,43 +1,21 @@
import re
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import Final
from urllib.parse import unquote
import pytest
from integration._support.upstream import _aws_event_frame
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue
import litellm
from litellm.exceptions import BadGatewayError, BadRequestError, MidStreamFallbackError
from litellm.exceptions import BadRequestError
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
_MODEL_ID: Final = "global.moonshotai.kimi-k3"
_CONVERSE_MODEL: Final = f"bedrock/converse/{_MODEL_ID}"
_STREAM_TARGET: Final = f"/model/{_MODEL_ID}/converse-stream"
_EVENT_STREAM: Final = "application/vnd.amazon.eventstream"
_ANSWER: Final = "bedrock sync decoder control"
_REJECTION: Final = "structured output schema uses unsupported regex negative look-ahead"
_UNKNOWN_TYPE: Final = "somethingBedrockAddedLater"
_USAGE: Final[dict[str, JsonValue]] = {"usage": {"inputTokens": 11, "outputTokens": 4, "totalTokens": 15}}
def _frame(event_type: str, payload: Mapping[str, JsonValue]) -> bytes:
return _aws_event_frame(event_type, payload, "sc", "u")
_NORMAL: Final = b"".join(
(
_frame("messageStart", {"role": "assistant"}),
_frame("contentBlockDelta", {"delta": {"text": _ANSWER}, "contentBlockIndex": 0}),
_frame("contentBlockStop", {"contentBlockIndex": 0}),
_frame("messageStop", {"stopReason": "end_turn"}),
_frame("metadata", _USAGE),
)
)
_VALIDATION_FRAME: Final = _frame("validationException", {"message": _REJECTION})
_UNKNOWN_FRAME: Final = _frame(_UNKNOWN_TYPE, {"future": True})
@dataclass(frozen=True, slots=True)
@ -46,14 +24,6 @@ class _Consumed:
finish_reasons: tuple[str | None, ...]
def _peer(frames: bytes) -> Callable[[Request], Reply]:
def respond(request: Request) -> Reply:
assert unquote(request.target) == _STREAM_TARGET, request.target
return Reply(body=frames, content_type=_EVENT_STREAM)
return respond
def _consume_sync_stream(wire: Wire) -> _Consumed:
response: Final = litellm.completion(
model=_CONVERSE_MODEL,
@ -74,32 +44,6 @@ def _consume_sync_stream(wire: Wire) -> _Consumed:
)
def test_k01_sync_stream_with_a_validation_exception_frame_first_raises_a_400() -> None:
with wire_server(_peer(_VALIDATION_FRAME)) as wire:
with pytest.raises(BadRequestError, match=re.escape(_REJECTION)) as raised:
_consume_sync_stream(wire)
assert raised.value.status_code == 400, raised.value
assert len(wire.drain()) == 1
def test_k02_sync_stream_whose_only_frame_has_an_unknown_event_type_raises_a_502_naming_it() -> None:
with wire_server(_peer(_UNKNOWN_FRAME)) as wire:
with pytest.raises(MidStreamFallbackError, match=re.escape(_UNKNOWN_TYPE)) as raised:
_consume_sync_stream(wire)
assert raised.value.status_code == 502, raised.value
assert isinstance(raised.value.original_exception, BadGatewayError), raised.value.original_exception
assert "none of its 1 events carried a known event type" in str(raised.value), raised.value
assert len(wire.drain()) == 1
def test_k03_sync_stream_with_normal_frames_delivers_the_text_and_a_stop() -> None:
with wire_server(_peer(_NORMAL)) as wire:
consumed: Final = _consume_sync_stream(wire)
assert consumed.text == _ANSWER, consumed
assert consumed.finish_reasons[-1] == "stop", consumed
assert len(wire.drain()) == 1
def test_k04_sync_stream_open_400_carries_the_bedrock_body_as_text() -> None:
body: Final = '{"message":"The request is invalid."}'

View file

@ -1,12 +1,13 @@
import copy
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 (
@ -6889,7 +6890,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",
@ -6904,9 +6918,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,

View file

@ -14,7 +14,6 @@ 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.litellm_core_utils.prompt_templates.mid_conversation_system import CONVERTED_SYSTEM_NOTE
from litellm.llms.anthropic.chat.transformation import (
@ -22,7 +21,6 @@ from litellm.llms.anthropic.chat.transformation import (
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
@ -4269,12 +4267,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."}],
@ -4282,7 +4285,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: