From ddd506492ef3276abb8a86a4c0f8295f7cb22329 Mon Sep 17 00:00:00 2001 From: narmstrong-berri Date: Fri, 16 Oct 2026 06:27:46 +0000 Subject: [PATCH] test(native): assert each cross-API bridge runs hooks, logging and the provider call once --- tests/unit/native/test_route_plan.py | 196 ++++++++++++++++++++++++++- 1 file changed, 194 insertions(+), 2 deletions(-) diff --git a/tests/unit/native/test_route_plan.py b/tests/unit/native/test_route_plan.py index d6c089f9e22..1db632fa30f 100644 --- a/tests/unit/native/test_route_plan.py +++ b/tests/unit/native/test_route_plan.py @@ -1,10 +1,19 @@ +import asyncio +import json +from collections.abc import AsyncIterable, Callable, Coroutine, Mapping, Sequence +from typing import Final, cast + +import httpx import pytest +import respx +from pydantic import TypeAdapter import litellm - +from litellm.integrations.custom_logger import CustomLogger from litellm.llms.anthropic.pass_through.messages.transformation import AnthropicMessagesConfig from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig from litellm.llms.openai_like.messages.transformation import OpenAILikeAnthropicMessagesConfig +from litellm.messages.dispatch import anthropic_messages from litellm.native.route_plan import ( NativeChat, NativeMessages, @@ -65,8 +74,15 @@ def test_messages_without_a_native_config_bridge_to_responses_only_for_openai( assert _plan(model, provider, **kwargs) == expected +_MODEL_COST: Final = TypeAdapter(dict[str, object]) + + +def _isolate_model_cost(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(litellm, "model_cost", _MODEL_COST.validate_python(litellm.model_cost)) + + def test_forced_chat_keeps_thinking_on_chat_for_a_model_that_cannot_reason(monkeypatch: pytest.MonkeyPatch): - monkeypatch.setattr(litellm, "model_cost", {**litellm.model_cost}) + _isolate_model_cost(monkeypatch) litellm.register_model({"openai/lit-plain-model": {"litellm_provider": "openai", "supports_reasoning": False}}) plan = _plan("lit-plain-model", "openai", force_chat_completions=True, thinking_enabled=True) @@ -109,3 +125,179 @@ def test_responses_reach_chat_without_a_provider_config_or_when_asked( ) def test_chat_bridges_to_responses_only_when_the_provider_serves_it(has_config: bool, expected: RoutePlan): assert plan_chat_route(OpenAIResponsesAPIConfig() if has_config else None) == expected + + +CHAT_BASE: Final = "http://chat-upstream.test/v1" +RESPONSES_BASE: Final = "http://responses-upstream.test/v1" +CHAT_MODEL: Final = "hosted_vllm/served-model" +RESPONSES_ONLY_MODEL: Final = "openai/bridge-once-model" +USER_TURN: Final = ({"role": "user", "content": "hi"},) + +_CHAT_REPLY: Final[Mapping[str, object]] = { + "id": "chatcmpl-once", + "object": "chat.completion", + "created": 1, + "model": "served-model", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "once"}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, +} +_CHAT_CHUNKS: Final[tuple[Mapping[str, object], ...]] = ( + { + "id": "chatcmpl-once", + "object": "chat.completion.chunk", + "created": 1, + "model": "served-model", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "once"}, "finish_reason": None}], + }, + { + "id": "chatcmpl-once", + "object": "chat.completion.chunk", + "created": 1, + "model": "served-model", + "choices": [{"index": 0, "delta": dict[str, object](), "finish_reason": "stop"}], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + }, +) +_OUTPUT_MESSAGE: Final[Mapping[str, object]] = { + "type": "message", + "id": "msg_once", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "once", "annotations": list[object]()}], +} +_RESPONSES_REPLY: Final[Mapping[str, object]] = { + "id": "resp_once", + "object": "response", + "created_at": 1, + "model": "bridge-once-model", + "status": "completed", + "output": [_OUTPUT_MESSAGE], + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": list[object](), + "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, +} +_RESPONSES_EVENTS: Final[tuple[Mapping[str, object], ...]] = ( + {"type": "response.created", "sequence_number": 0, "response": {**_RESPONSES_REPLY, "status": "in_progress"}}, + {"type": "response.output_item.added", "sequence_number": 1, "output_index": 0, "item": _OUTPUT_MESSAGE}, + { + "type": "response.output_text.delta", + "sequence_number": 2, + "item_id": "msg_once", + "output_index": 0, + "content_index": 0, + "delta": "once", + }, + {"type": "response.output_item.done", "sequence_number": 3, "output_index": 0, "item": _OUTPUT_MESSAGE}, + {"type": "response.completed", "sequence_number": 4, "response": _RESPONSES_REPLY}, +) + + +def _sse(events: Sequence[Mapping[str, object]], *, done: bool) -> bytes: + frames: Final = "".join(f"data: {json.dumps(event)}\n\n" for event in events) + return (frames + ("data: [DONE]\n\n" if done else "")).encode() + + +class _CallCounter(CustomLogger): + pre_call: int = 0 + success: int = 0 + + async def async_pre_call_deployment_hook(self, kwargs: Mapping[str, object], call_type: object) -> None: + self.pre_call += 1 + + async def async_log_success_event( + self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + self.success += 1 + + +async def _messages(model: str, api_base: str, stream: bool) -> object: + return await anthropic_messages( + model=model, api_base=api_base, api_key="sk-x", max_tokens=10, messages=list(USER_TURN), stream=stream + ) + + +async def _messages_to_chat(stream: bool) -> object: + return await _messages(CHAT_MODEL, CHAT_BASE, stream) + + +async def _messages_to_responses(stream: bool) -> object: + return await _messages("openai/gpt-5.6-luna", RESPONSES_BASE, stream) + + +async def _chat_to_responses(stream: bool) -> object: + return await litellm.acompletion( + model=RESPONSES_ONLY_MODEL, api_base=RESPONSES_BASE, api_key="sk-x", messages=list(USER_TURN), stream=stream + ) + + +async def _responses_to_chat(stream: bool) -> object: + return await litellm.aresponses( + model="deepseek/served-model", api_base=CHAT_BASE, api_key="sk-x", input="hi", stream=stream + ) + + +async def _drained(reply: object) -> None: + if isinstance(reply, AsyncIterable): + async for _ in cast(AsyncIterable[object], reply): # cast-ok: every bridged stream is drained the same way + pass + + +async def _settled(counter: _CallCounter) -> _CallCounter: + for _ in range(200): + if counter.success: + break + await asyncio.sleep(0.01) + await asyncio.sleep(0.1) + return counter + + +_BRIDGES: Final = ( + pytest.param(_messages_to_chat, "chat", id="messages_to_chat"), + pytest.param(_messages_to_responses, "responses", id="messages_to_responses"), + pytest.param(_chat_to_responses, "responses", id="chat_to_responses"), + pytest.param(_responses_to_chat, "chat", id="responses_to_chat"), +) + + +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize(("bridge", "upstream_api"), _BRIDGES) +@pytest.mark.asyncio +async def test_a_bridged_request_runs_hooks_logs_and_calls_the_provider_once( + bridge: Callable[[bool], Coroutine[object, object, object]], + upstream_api: str, + stream: bool, + respx_mock: respx.MockRouter, + monkeypatch: pytest.MonkeyPatch, +): + counter: Final = _CallCounter() + _isolate_model_cost(monkeypatch) + monkeypatch.setattr(litellm, "callbacks", [counter]) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + litellm.in_memory_llm_clients_cache.flush_cache() + litellm.register_model( + {RESPONSES_ONLY_MODEL: {"litellm_provider": "openai", "mode": "responses", "max_tokens": 100}} + ) + upstream: Final = ( + respx_mock.post(f"{CHAT_BASE}/chat/completions").mock( + return_value=httpx.Response( + 200, headers={"content-type": "text/event-stream"}, content=_sse(_CHAT_CHUNKS, done=True) + ) + if stream + else httpx.Response(200, json=_CHAT_REPLY) + ) + if upstream_api == "chat" + else respx_mock.post(f"{RESPONSES_BASE}/responses").mock( + return_value=httpx.Response( + 200, headers={"content-type": "text/event-stream"}, content=_sse(_RESPONSES_EVENTS, done=False) + ) + if stream + else httpx.Response(200, json=_RESPONSES_REPLY) + ) + ) + + await _drained(await bridge(stream)) + settled: Final = await _settled(counter) + + assert upstream.call_count == 1 + assert (settled.pre_call, settled.success) == (1, 1)