scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-11 08:38:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 359

test: type the developer-hoist tests and fake HTTP with respx

Browse source
This commit is contained in:
mateo-berri 2026-10-28 15:47:47 -07:00
parent 7c4ef7c30d
commit 8cb3248563
11 changed files with 197 additions and 161 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
tests/unit/llms/azure/chat/test_azure_chat_gpt_transformation.py
View file

@ -796,8 +796,8 @@ def _developer_message_after_a_user_turn() -> list[dict[str, str]]:
@pytest.mark.parametrize("config", [AzureOpenAIConfig(), AzureOpenAIGPT5Config()])
def test_azure_keeps_a_later_developer_message_in_place_as_system(config):
messages = config.translate_developer_role_to_system_role(
def test_azure_keeps_a_later_developer_message_in_place_as_system(config: AzureOpenAIConfig) -> None:
messages: Final = config.translate_developer_role_to_system_role(
messages=_developer_message_after_a_user_turn(),
model="my-deployment",
custom_llm_provider="azure",

7
tests/unit/llms/azure_ai/chat/test_azure_ai_transformation.py
View file

@ -1,5 +1,6 @@
import json
import traceback
from collections.abc import Callable
from typing import Final
from unittest.mock import MagicMock, patch
@ -1101,10 +1102,10 @@ def test_map_azure_model_group(model_group_header, expected_model):
def test_translate_developer_role_hoists_a_later_developer_message_for_a_model_that_rejects_it(
declare_mid_conversation_system,
):
declare_mid_conversation_system: Callable[[str, bool], None],
) -> None:
declare_mid_conversation_system("azure_ai/system-first-model", False)
messages = AzureAIStudioConfig().translate_developer_role_to_system_role(
messages: Final = AzureAIStudioConfig().translate_developer_role_to_system_role(
messages=[
{"role": "system", "content": "You are terse."},
{"role": "user", "content": "Hi there"},

132
tests/unit/llms/base_llm/test_base_utils.py
View file

@ -1,3 +1,5 @@
from typing import Final
from litellm.llms.base_llm.base_utils import (
hoist_developer_messages_into_leading_system_message,
merge_system_message_runs,
@ -5,8 +7,8 @@ from litellm.llms.base_llm.base_utils import (
class TestMergeSystemMessageRuns:
def test_later_developer_message_stays_in_place_while_the_leading_run_folds(self):
messages = [
def test_later_developer_message_stays_in_place_while_the_leading_run_folds(self) -> None:
messages: Final = [
{"role": "system", "content": "You are Codex"},
{"role": "developer", "content": "Sandbox: read-only."},
{"role": "user", "content": "Hi there"},
@ -20,8 +22,8 @@ class TestMergeSystemMessageRuns:
{"role": "user", "content": "What is the capital of France?"},
]
def test_run_with_content_that_is_neither_text_nor_blocks_is_forwarded_unmerged(self):
messages = [
def test_run_with_content_that_is_neither_text_nor_blocks_is_forwarded_unmerged(self) -> None:
messages: Final = [
{"role": "system", "content": "You are Codex"},
{"role": "developer", "content": {"text": "not a content list"}},
{"role": "user", "content": "Hi there"},
@ -32,8 +34,8 @@ class TestMergeSystemMessageRuns:
{"role": "user", "content": "Hi there"},
]
def test_hoisted_developer_message_with_content_that_is_neither_text_nor_blocks_is_forwarded_unmerged(self):
messages = [
def test_hoisted_developer_message_with_content_that_is_neither_text_nor_blocks_is_forwarded_unmerged(self) -> None:
messages: Final = [
{"role": "system", "content": "You are Codex"},
{"role": "user", "content": "Hi there"},
{"role": "developer", "content": 42},
@ -48,8 +50,8 @@ class TestMergeSystemMessageRuns:
class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
def test_single_developer_message(self):
messages = [
def test_single_developer_message(self) -> None:
messages: Final = [
{"role": "developer", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello"},
]
@ -58,8 +60,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Hello"},
]
def test_consecutive_developer_messages_merge(self):
messages = [
def test_consecutive_developer_messages_merge(self) -> None:
messages: Final = [
{"role": "developer", "content": "Rule 1: Be concise."},
{"role": "developer", "content": "Rule 2: Respond in Markdown."},
{"role": "user", "content": "Hello"},
@ -69,8 +71,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Hello"},
]
def test_system_followed_by_developer_merges(self):
messages = [
def test_system_followed_by_developer_merges(self) -> None:
messages: Final = [
{"role": "system", "content": "Base instructions."},
{"role": "developer", "content": "Developer override."},
{"role": "user", "content": "Hello"},
@ -80,8 +82,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Hello"},
]
def test_developer_followed_by_system_merges(self):
messages = [
def test_developer_followed_by_system_merges(self) -> None:
messages: Final = [
{"role": "developer", "content": "Developer instruction."},
{"role": "system", "content": "System instruction."},
{"role": "user", "content": "Hello"},
@ -91,8 +93,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Hello"},
]
def test_consecutive_native_system_messages_merge(self):
messages = [
def test_consecutive_native_system_messages_merge(self) -> None:
messages: Final = [
{"role": "system", "content": "System part 1."},
{"role": "system", "content": "System part 2."},
{"role": "user", "content": "Hello"},
@ -102,7 +104,7 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Hello"},
]
def test_empty_string_side_does_not_pad(self):
def test_empty_string_side_does_not_pad(self) -> None:
assert list(
hoist_developer_messages_into_leading_system_message(
[{"role": "system", "content": ""}, {"role": "developer", "content": "Rules"}]
@ -114,8 +116,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
)
) == [{"role": "system", "content": "Rules"}]
def test_empty_string_message_adds_no_block_when_the_run_merges_as_blocks(self):
messages = [
def test_empty_string_message_adds_no_block_when_the_run_merges_as_blocks(self) -> None:
messages: Final = [
{"role": "system", "content": ""},
{"role": "developer", "content": [{"type": "text", "text": "Rule 1"}, {"type": "text", "text": "Rule 2"}]},
{"role": "user", "content": "Hi"},
@ -125,8 +127,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Hi"},
]
def test_two_cached_string_messages_keep_one_breakpoint_each(self):
messages = [
def test_two_cached_string_messages_keep_one_breakpoint_each(self) -> None:
messages: Final = [
{"role": "system", "content": "A", "cache_control": {"type": "ephemeral", "ttl": "5m"}},
{"role": "system", "content": "B", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
]
@ -140,8 +142,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
},
]
def test_cache_control_supplied_only_by_the_later_message_marks_only_its_block(self):
messages = [
def test_cache_control_supplied_only_by_the_later_message_marks_only_its_block(self) -> None:
messages: Final = [
{"role": "system", "content": "A"},
{"role": "developer", "content": "B", "cache_control": {"type": "ephemeral"}},
]
@ -155,8 +157,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
},
]
def test_cache_control_of_the_first_message_stays_on_its_own_block(self):
messages = [
def test_cache_control_of_the_first_message_stays_on_its_own_block(self) -> None:
messages: Final = [
{"role": "system", "content": "Cached prompt", "cache_control": {"type": "ephemeral"}},
{"role": "developer", "content": "Additional instructions"},
{"role": "user", "content": "Hello"},
@ -172,8 +174,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Hello"},
]
def test_cached_string_message_merged_with_block_content_keeps_its_breakpoint(self):
messages = [
def test_cached_string_message_merged_with_block_content_keeps_its_breakpoint(self) -> None:
messages: Final = [
{"role": "system", "content": "Cached prompt", "cache_control": {"type": "ephemeral"}},
{"role": "user", "content": "Hi there"},
{"role": "developer", "content": [{"type": "text", "text": "Answer with exactly one word."}]},
@ -191,8 +193,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "What is the capital of France?"},
]
def test_plain_messages_ahead_of_a_cached_one_keep_their_own_blocks(self):
messages = [
def test_plain_messages_ahead_of_a_cached_one_keep_their_own_blocks(self) -> None:
messages: Final = [
{"role": "system", "content": "Base"},
{"role": "developer", "content": "Rule"},
{"role": "developer", "content": "Cached", "cache_control": {"type": "ephemeral"}},
@ -210,21 +212,21 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Hello"},
]
def test_name_of_the_last_named_message_survives_a_long_run(self):
messages = [
def test_name_of_the_last_named_message_survives_a_long_run(self) -> None:
messages: Final = [
{"role": "system", "content": "Base", "name": "first"},
*({"role": "developer", "content": f"Rule {index}"} for index in range(2000)),
{"role": "developer", "content": "Last", "name": "last"},
{"role": "user", "content": "Hello"},
]
merged = list(hoist_developer_messages_into_leading_system_message(messages))
merged: Final = list(hoist_developer_messages_into_leading_system_message(messages))
assert len(merged) == 2
assert merged[0]["name"] == "last"
assert merged[0]["content"] == "\n\n".join(["Base", *(f"Rule {index}" for index in range(2000)), "Last"])
def test_billing_metadata_keeps_its_own_block_when_merged(self):
billing = {"role": "system", "content": "x-anthropic-billing-header: cc_version=1.0;"}
advisory = {"role": "system", "content": "Guardrail advisory: request was flagged"}
def test_billing_metadata_keeps_its_own_block_when_merged(self) -> None:
billing: Final = {"role": "system", "content": "x-anthropic-billing-header: cc_version=1.0;"}
advisory: Final = {"role": "system", "content": "Guardrail advisory: request was flagged"}
assert list(hoist_developer_messages_into_leading_system_message([billing, advisory])) == [
{
"role": "system",
@ -244,8 +246,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
}
]
def test_developer_message_after_a_user_turn_is_hoisted_into_the_leading_system_message(self):
messages = [
def test_developer_message_after_a_user_turn_is_hoisted_into_the_leading_system_message(self) -> None:
messages: Final = [
{"role": "system", "content": "System turn 1"},
{"role": "user", "content": "User turn 1"},
{"role": "developer", "content": "Developer turn 2"},
@ -257,8 +259,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "User turn 2"},
]
def test_developer_message_in_second_position_without_a_leading_system_message_is_hoisted(self):
messages = [
def test_developer_message_in_second_position_without_a_leading_system_message_is_hoisted(self) -> None:
messages: Final = [
{"role": "user", "content": "Hi there"},
{"role": "developer", "content": "Answer with exactly one word."},
{"role": "user", "content": "What is the capital of France?"},
@ -269,8 +271,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "What is the capital of France?"},
]
def test_every_later_developer_message_is_hoisted_in_order_across_assistant_and_tool_turns(self):
messages = [
def test_every_later_developer_message_is_hoisted_in_order_across_assistant_and_tool_turns(self) -> None:
messages: Final = [
{"role": "system", "content": "Base"},
{"role": "user", "content": "Turn 1"},
{"role": "assistant", "content": "Reply 1"},
@ -287,8 +289,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Turn 3"},
]
def test_developer_message_that_closes_the_conversation_after_an_assistant_turn_becomes_a_user_turn(self):
messages = [
def test_developer_message_that_closes_the_conversation_after_an_assistant_turn_becomes_a_user_turn(self) -> None:
messages: Final = [
{"role": "system", "content": "You are terse."},
{"role": "user", "content": "Hi there"},
{"role": "assistant", "content": "Hello! How can I help?"},
@ -301,8 +303,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Reply with the single word PONG and nothing else."},
]
def test_closing_developer_message_keeps_its_cache_mark_as_a_user_turn(self):
messages = [
def test_closing_developer_message_keeps_its_cache_mark_as_a_user_turn(self) -> None:
messages: Final = [
{"role": "user", "content": "Hi there"},
{"role": "assistant", "content": "Hello! How can I help?"},
{"role": "developer", "content": "Reply with PONG.", "cache_control": {"type": "ephemeral"}},
@ -313,8 +315,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
"cache_control": {"type": "ephemeral"},
}
def test_only_the_closing_developer_run_after_an_assistant_turn_becomes_user_turns(self):
messages = [
def test_only_the_closing_developer_run_after_an_assistant_turn_becomes_user_turns(self) -> None:
messages: Final = [
{"role": "system", "content": "Base"},
{"role": "user", "content": "Turn 1"},
{"role": "developer", "content": "Update A"},
@ -332,8 +334,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Closing C"},
]
def test_developer_message_that_closes_the_conversation_after_a_user_turn_is_hoisted(self):
messages = [
def test_developer_message_that_closes_the_conversation_after_a_user_turn_is_hoisted(self) -> None:
messages: Final = [
{"role": "user", "content": "Hi there"},
{"role": "developer", "content": "Reply with exactly one word: the capital of France"},
]
@ -342,8 +344,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "Hi there"},
]
def test_developer_message_that_closes_the_conversation_after_a_tool_result_is_hoisted(self):
messages = [
def test_developer_message_that_closes_the_conversation_after_a_tool_result_is_hoisted(self) -> None:
messages: Final = [
{"role": "system", "content": "Base"},
{"role": "user", "content": "Look it up"},
{
@ -365,8 +367,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "tool", "tool_call_id": "call_1", "content": "Paris"},
]
def test_hoisted_single_text_block_developer_message_joins_string_instructions_as_text(self):
messages = [
def test_hoisted_single_text_block_developer_message_joins_string_instructions_as_text(self) -> None:
messages: Final = [
{"role": "system", "content": "You are Codex"},
{"role": "user", "content": [{"type": "text", "text": "# AGENTS.md"}]},
{"role": "developer", "content": [{"type": "text", "text": "<permissions instructions>"}]},
@ -378,8 +380,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": [{"type": "text", "text": "What is the capital of France?"}]},
]
def test_multi_block_developer_message_keeps_its_blocks_after_string_instructions(self):
messages = [
def test_multi_block_developer_message_keeps_its_blocks_after_string_instructions(self) -> None:
messages: Final = [
{"role": "system", "content": "You are Codex"},
{"role": "user", "content": "Hi"},
{"role": "developer", "content": [{"type": "text", "text": "Rule 1"}, {"type": "text", "text": "Rule 2"}]},
@ -398,8 +400,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "What is the capital of France?"},
]
def test_system_message_without_content_folds_into_the_run_instead_of_raising(self):
messages = [
def test_system_message_without_content_folds_into_the_run_instead_of_raising(self) -> None:
messages: Final = [
{"role": "system", "content": "Rules"},
{"role": "system"},
{"role": "system", "content": None},
@ -410,8 +412,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "What is the capital of France?"},
]
def test_message_level_cache_control_of_a_block_content_member_lands_on_its_last_block(self):
messages = [
def test_message_level_cache_control_of_a_block_content_member_lands_on_its_last_block(self) -> None:
messages: Final = [
{
"role": "system",
"content": [{"type": "text", "text": "Rule 1"}, {"type": "text", "text": "Rule 2"}],
@ -432,8 +434,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "What is the capital of France?"},
]
def test_message_level_cache_control_does_not_override_a_block_that_already_carries_one(self):
messages = [
def test_message_level_cache_control_does_not_override_a_block_that_already_carries_one(self) -> None:
messages: Final = [
{
"role": "system",
"content": [{"type": "text", "text": "Rule 1", "cache_control": {"type": "ephemeral", "ttl": "1h"}}],
@ -453,8 +455,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
{"role": "user", "content": "What is the capital of France?"},
]
def test_non_consecutive_native_system_messages_stay_where_the_client_put_them(self):
messages = [
def test_non_consecutive_native_system_messages_stay_where_the_client_put_them(self) -> None:
messages: Final = [
{"role": "system", "content": "System turn 1"},
{"role": "user", "content": "User turn 1"},
{"role": "system", "content": "System turn 2"},
@ -462,8 +464,8 @@ class TestHoistDeveloperMessagesIntoLeadingSystemMessage:
]
assert list(hoist_developer_messages_into_leading_system_message(messages)) == messages
def test_list_content_merge_preserves_block_level_cache_control(self):
messages = [
def test_list_content_merge_preserves_block_level_cache_control(self) -> None:
messages: Final = [
{"role": "system", "content": [{"type": "text", "text": "Cached", "cache_control": {"type": "ephemeral"}}]},
{"role": "developer", "content": [{"type": "text", "text": "Extra"}]},
{"role": "user", "content": "Hello"},

21
tests/unit/llms/fireworks_ai/chat/test_fireworks_ai_chat_transformation.py
View file

@ -1,4 +1,5 @@
import json
from collections.abc import Callable
from typing import Final
from unittest.mock import MagicMock, patch
@ -2368,8 +2369,8 @@ _CODEX_DEVELOPER_MESSAGE_AFTER_A_USER_TURN: Final = (
)
def test_translate_developer_role_keeps_a_later_developer_message_in_place_for_an_undeclared_fireworks_model():
messages = FireworksAIConfig().translate_developer_role_to_system_role(
def test_translate_developer_role_keeps_a_later_developer_message_in_place_for_an_undeclared_fireworks_model() -> None:
messages: Final = FireworksAIConfig().translate_developer_role_to_system_role(
messages=_CODEX_DEVELOPER_MESSAGE_AFTER_A_USER_TURN,
model="accounts/fireworks/models/unlisted-model",
custom_llm_provider="fireworks_ai",
@ -2385,8 +2386,8 @@ def test_translate_developer_role_keeps_a_later_developer_message_in_place_for_a
]
def test_translate_developer_role_leaves_a_leading_system_and_developer_pair_unmerged_for_an_undeclared_model():
messages = FireworksAIConfig().translate_developer_role_to_system_role(
def test_translate_developer_role_leaves_a_leading_system_and_developer_pair_unmerged_for_an_undeclared_model() -> None:
messages: Final = FireworksAIConfig().translate_developer_role_to_system_role(
messages=[
{"role": "system", "content": "You are Codex"},
{"role": "developer", "content": "Answer with exactly one word."},
@ -2407,9 +2408,9 @@ def test_translate_developer_role_leaves_a_leading_system_and_developer_pair_unm
@pytest.mark.parametrize("model", ["qwen3p8-max", "accounts/fireworks/models/qwen3p8-max"])
def test_translate_developer_role_hoists_a_later_developer_message_for_the_system_first_qwen3p8_model(
local_model_cost_map, model
):
messages = FireworksAIConfig().translate_developer_role_to_system_role(
local_model_cost_map: None, model: str
) -> None:
messages: Final = FireworksAIConfig().translate_developer_role_to_system_role(
messages=_CODEX_DEVELOPER_MESSAGE_AFTER_A_USER_TURN,
model=model,
custom_llm_provider="fireworks_ai",
@ -2429,10 +2430,10 @@ def test_translate_developer_role_hoists_a_later_developer_message_for_the_syste
[(False, ["system", "user", "user"]), (True, ["system", "user", "system", "user"])],
)
def test_translate_developer_role_follows_the_deployment_model_info_over_the_cost_map(
declare_mid_conversation_system, deployment_declared, expected_roles
):
declare_mid_conversation_system: Callable[[str, bool], None], deployment_declared: bool, expected_roles: list[str]
) -> None:
declare_mid_conversation_system("fireworks_ai/accounts/fireworks/models/shared-backend", not deployment_declared)
messages = FireworksAIConfig().translate_developer_role_to_system_role(
messages: Final = FireworksAIConfig().translate_developer_role_to_system_role(
messages=_CODEX_DEVELOPER_MESSAGE_AFTER_A_USER_TURN,
model="accounts/fireworks/models/shared-backend",
custom_llm_provider="fireworks_ai",

69
tests/unit/llms/fireworks_ai/responses/test_fireworks_ai_responses_transformation.py
View file

@ -1,5 +1,5 @@
import json
from collections.abc import Mapping
from collections.abc import Callable, Mapping
from types import MappingProxyType
from typing import Final, TypedDict, cast
from unittest.mock import MagicMock, patch
@ -7,6 +7,7 @@ from urllib.parse import quote
import httpx
import pytest
import respx
from openai.types.responses import (
EasyInputMessage,
ResponseFunctionToolCall,
@ -81,6 +82,14 @@ def _mock_http_client(response_body: Mapping[str, object]) -> MagicMock:
return client
def _fireworks_responses_route(model: str) -> respx.Route:
return respx.post(FIREWORKS_RESPONSES_URL).mock(return_value=httpx.Response(200, json=_fireworks_response(model)))
def _routed_body(route: respx.Route) -> Mapping[str, object]:
return json.loads(route.calls.last.request.content)
def _sent_request(client: MagicMock) -> tuple[str, Mapping[str, str], Mapping[str, object]]:
kwargs: Final = client.post.call_args.kwargs
body: Final = kwargs["json"] if "json" in kwargs else json.loads(kwargs["data"])
@ -182,23 +191,23 @@ def test_responses_call_folds_developer_items_into_instructions() -> None:
)
@respx.mock
def test_responses_call_keeps_developer_items_in_place_for_a_model_that_accepts_later_system_items(
declare_mid_conversation_system,
declare_mid_conversation_system: Callable[[str, bool], None],
) -> None:
declare_mid_conversation_system("fireworks_ai/accounts/fireworks/models/accepts-later-system", True)
client: Final = _mock_http_client(_fireworks_response("accounts/fireworks/models/accepts-later-system"))
with patch(HTTPX_CLIENT_FACTORY, return_value=client):
litellm.responses(
model="fireworks_ai/accounts/fireworks/models/accepts-later-system",
instructions="You are Codex",
input=[
{"role": "user", "content": "Hi there"},
{"role": "developer", "content": "Answer with exactly one word."},
{"role": "user", "content": "What is the capital of France?"},
],
api_key="fw-test-key",
)
_, _, body = _sent_request(client)
route: Final = _fireworks_responses_route("accounts/fireworks/models/accepts-later-system")
litellm.responses(
model="fireworks_ai/accounts/fireworks/models/accepts-later-system",
instructions="You are Codex",
input=[
{"role": "user", "content": "Hi there"},
{"role": "developer", "content": "Answer with exactly one word."},
{"role": "user", "content": "What is the capital of France?"},
],
api_key="fw-test-key",
)
body: Final = _routed_body(route)
assert body["instructions"] == "You are Codex"
assert tuple(body["input"]) == (
{"role": "user", "content": "Hi there"},
@ -207,6 +216,7 @@ def test_responses_call_keeps_developer_items_in_place_for_a_model_that_accepts_
)
@respx.mock
@pytest.mark.parametrize(
("model_info", "closing_role"),
[(None, "system"), ({"id": "deployment-1", "supports_mid_conversation_system": False}, "user")],
@ -214,21 +224,20 @@ def test_responses_call_keeps_developer_items_in_place_for_a_model_that_accepts_
def test_responses_call_sends_the_closing_developer_item_as_a_user_turn_only_for_a_declared_system_first_deployment(
model_info: Mapping[str, object] | None, closing_role: str
) -> None:
client: Final = _mock_http_client(_fireworks_response("accounts/fireworks/models/kimi-k3"))
with patch(HTTPX_CLIENT_FACTORY, return_value=client):
litellm.responses(
model="fireworks_ai/accounts/fireworks/models/kimi-k3",
input=[
{"role": "user", "content": "Hi there"},
{"role": "developer", "content": "Answer with exactly one word."},
{"role": "user", "content": "What is the capital of France?"},
{"type": "message", "role": "assistant", "content": "Paris"},
{"role": "developer", "content": "Reply with PONG."},
],
model_info=model_info,
api_key="fw-test-key",
)
_, _, body = _sent_request(client)
route: Final = _fireworks_responses_route("accounts/fireworks/models/kimi-k3")
litellm.responses(
model="fireworks_ai/accounts/fireworks/models/kimi-k3",
input=[
{"role": "user", "content": "Hi there"},
{"role": "developer", "content": "Answer with exactly one word."},
{"role": "user", "content": "What is the capital of France?"},
{"type": "message", "role": "assistant", "content": "Paris"},
{"role": "developer", "content": "Reply with PONG."},
],
model_info=model_info,
api_key="fw-test-key",
)
body: Final = _routed_body(route)
assert body["instructions"] == "Answer with exactly one word."
assert tuple(body["input"]) == (
{"role": "user", "content": "Hi there"},

82
tests/unit/llms/hosted_vllm/responses/test_hosted_vllm_responses.py
View file

@ -8,11 +8,13 @@ hosted_vllm (and any OpenAI-compatible provider using add_provider_specific_para
"""
import json
from collections.abc import Mapping
from collections.abc import Callable, Mapping
from typing import Final
from unittest.mock import MagicMock, patch
import httpx
import pytest
import respx
import litellm
@ -194,40 +196,48 @@ _LATER_DEVELOPER_INPUT: Final = (
)
def _sent_body(mock_client: MagicMock) -> Mapping[str, object]:
kwargs: Final = mock_client.post.call_args.kwargs
return kwargs["json"] if "json" in kwargs else json.loads(kwargs["data"])
_VLLM_RESPONSES_URL: Final = "https://test-vllm.example.com/v1/responses"
def _vllm_responses_route(output_text: str) -> respx.Route:
return respx.post(_VLLM_RESPONSES_URL).mock(
return_value=httpx.Response(200, json=_make_mock_responses_api_response(output_text))
)
def _sent_body(route: respx.Route) -> Mapping[str, object]:
return json.loads(route.calls.last.request.content)
@respx.mock
def test_hosted_vllm_responses_folds_a_later_developer_item_into_instructions_for_a_system_first_model(
declare_mid_conversation_system,
):
declare_mid_conversation_system: Callable[[str, bool], None],
) -> None:
declare_mid_conversation_system("hosted_vllm/Qwen/Qwen3.6-27B", False)
mock_client: Final = _make_mock_http_client(_make_mock_responses_api_response("Paris"))
with patch("litellm.llms.custom_httpx.llm_http_handler.get_httpx_client", return_value=mock_client):
litellm.responses(
model="hosted_vllm/Qwen/Qwen3.6-27B",
instructions="You are a terse assistant.",
input=list(_LATER_DEVELOPER_INPUT),
api_base="https://test-vllm.example.com/v1",
)
body: Final = _sent_body(mock_client)
route: Final = _vllm_responses_route("Paris")
litellm.responses(
model="hosted_vllm/Qwen/Qwen3.6-27B",
instructions="You are a terse assistant.",
input=list(_LATER_DEVELOPER_INPUT),
api_base="https://test-vllm.example.com/v1",
)
body: Final = _sent_body(route)
assert body["instructions"] == "You are a terse assistant.\n\nAnswer with exactly one word."
assert tuple(body["input"]) == (_LATER_DEVELOPER_INPUT[0], _LATER_DEVELOPER_INPUT[2])
def test_hosted_vllm_responses_folds_for_a_deployment_declared_system_first_and_sends_a_closing_developer_as_user():
@respx.mock
def test_hosted_vllm_responses_folds_for_a_deployment_declared_system_first_and_sends_a_closing_developer_as_user() -> None:
assistant_turn: Final = {"type": "message", "role": "assistant", "content": "Paris"}
mock_client: Final = _make_mock_http_client(_make_mock_responses_api_response("PONG"))
with patch("litellm.llms.custom_httpx.llm_http_handler.get_httpx_client", return_value=mock_client):
litellm.responses(
model="hosted_vllm/Qwen/Qwen3-8B",
instructions="You are a terse assistant.",
input=[*_LATER_DEVELOPER_INPUT, assistant_turn, {"role": "developer", "content": "Reply with PONG."}],
model_info={"id": "deployment-1", "supports_mid_conversation_system": False},
api_base="https://test-vllm.example.com/v1",
)
body: Final = _sent_body(mock_client)
route: Final = _vllm_responses_route("PONG")
litellm.responses(
model="hosted_vllm/Qwen/Qwen3-8B",
instructions="You are a terse assistant.",
input=[*_LATER_DEVELOPER_INPUT, assistant_turn, {"role": "developer", "content": "Reply with PONG."}],
model_info={"id": "deployment-1", "supports_mid_conversation_system": False},
api_base="https://test-vllm.example.com/v1",
)
body: Final = _sent_body(route)
assert body["instructions"] == "You are a terse assistant.\n\nAnswer with exactly one word."
assert tuple(body["input"]) == (
_LATER_DEVELOPER_INPUT[0],
@ -237,15 +247,15 @@ def test_hosted_vllm_responses_folds_for_a_deployment_declared_system_first_and_
)
def test_hosted_vllm_responses_forwards_a_later_developer_item_in_place_for_an_undeclared_model():
mock_client: Final = _make_mock_http_client(_make_mock_responses_api_response("Paris"))
with patch("litellm.llms.custom_httpx.llm_http_handler.get_httpx_client", return_value=mock_client):
litellm.responses(
model="hosted_vllm/Qwen/Qwen3-8B",
instructions="You are a terse assistant.",
input=list(_LATER_DEVELOPER_INPUT),
api_base="https://test-vllm.example.com/v1",
)
body: Final = _sent_body(mock_client)
@respx.mock
def test_hosted_vllm_responses_forwards_a_later_developer_item_in_place_for_an_undeclared_model() -> None:
route: Final = _vllm_responses_route("Paris")
litellm.responses(
model="hosted_vllm/Qwen/Qwen3-8B",
instructions="You are a terse assistant.",
input=list(_LATER_DEVELOPER_INPUT),
api_base="https://test-vllm.example.com/v1",
)
body: Final = _sent_body(route)
assert body["instructions"] == "You are a terse assistant."
assert tuple(body["input"]) == _LATER_DEVELOPER_INPUT

4
tests/unit/llms/litellm_proxy/chat/test_litellm_proxy_chat_transformation.py
View file

@ -641,8 +641,8 @@ def test_litellm_gateway_from_sdk_with_thinking_param(respx_mock):
assert "Connection error." in str(e)
def test_translate_developer_role_leaves_a_later_developer_message_in_place_for_the_downstream_proxy():
translated = LiteLLMProxyChatConfig().translate_developer_role_to_system_role(
def test_translate_developer_role_leaves_a_later_developer_message_in_place_for_the_downstream_proxy() -> None:
translated: Final = LiteLLMProxyChatConfig().translate_developer_role_to_system_role(
messages=[
{"role": "system", "content": "You are terse."},
{"role": "user", "content": "Hi there"},

12
tests/unit/llms/openai/chat/test_openai_gpt_transformation.py
View file

@ -3,6 +3,7 @@ Tests for OpenAI GPT transformation (litellm/llms/openai/chat/gpt_transformation
"""
import base64
import json
from collections.abc import Callable
from typing import Final
import pytest
@ -1844,11 +1845,16 @@ def _developer_message_after_a_user_turn() -> list[dict[str, str]]:
],
)
def test_translate_developer_role_hoists_a_later_developer_message_only_for_a_model_off_openai_that_rejects_it(
declare_mid_conversation_system, config, custom_llm_provider, api_base, declared, expected_roles
):
declare_mid_conversation_system: Callable[[str, bool], None],
config: OpenAIGPTConfig,
custom_llm_provider: str,
api_base: str | None,
declared: bool | None,
expected_roles: list[str],
) -> None:
if declared is not None:
declare_mid_conversation_system(f"{custom_llm_provider}/system-first-model", declared)
messages = config.translate_developer_role_to_system_role(
messages: Final = config.translate_developer_role_to_system_role(
messages=_developer_message_after_a_user_turn(),
model="system-first-model",
custom_llm_provider=custom_llm_provider,

4
tests/unit/llms/openai/test_o_series_transformation.py
View file

@ -183,8 +183,8 @@ def test_litellm_responses():
assert isinstance(response.usage.completion_tokens_details, CompletionTokensDetails)
def test_translate_developer_role_keeps_developer_messages_untouched():
messages = [
def test_translate_developer_role_keeps_developer_messages_untouched() -> None:
messages: Final = [
{"role": "developer", "content": "Answer with exactly one word."},
{"role": "user", "content": "What is the capital of France?"},
]

12
tests/unit/llms/openai/test_openai.py
View file

@ -2,7 +2,7 @@ import asyncio
import importlib
import json
import os
from collections.abc import Mapping
from collections.abc import Callable, Mapping
from types import ModuleType
from typing import Final, Literal, Protocol
from unittest.mock import MagicMock, Mock, patch
@ -2032,11 +2032,15 @@ def _developer_message_after_a_user_turn() -> list[dict[str, str]]:
],
)
def test_openai_config_hoists_a_later_developer_message_only_for_a_model_off_openai_that_rejects_it(
declare_mid_conversation_system, custom_llm_provider, api_base, declared, expected_roles
):
declare_mid_conversation_system: Callable[[str, bool], None],
custom_llm_provider: str,
api_base: str,
declared: bool | None,
expected_roles: list[str],
) -> None:
if declared is not None:
declare_mid_conversation_system(f"{custom_llm_provider}/system-first-model", declared)
messages = OpenAIConfig().translate_developer_role_to_system_role(
messages: Final = OpenAIConfig().translate_developer_role_to_system_role(
messages=_developer_message_after_a_user_turn(),
model="system-first-model",
custom_llm_provider=custom_llm_provider,

11
tests/unit/test_utils.py
View file

@ -10879,11 +10879,14 @@ def test_get_optional_params_keeps_a_tool_config_cache_point_out_of_the_openai_e
],
)
def test_declared_mid_conversation_system_support_prefers_the_deployment_over_the_cost_map(
declare_mid_conversation_system, cost_map_declared, deployment_declared, expected
):
declare_mid_conversation_system: Callable[[str, bool], None],
cost_map_declared: bool | None,
deployment_declared: bool | None,
expected: bool | None,
) -> None:
if cost_map_declared is not None:
declare_mid_conversation_system("hosted_vllm/mid-conversation-probe", cost_map_declared)
deployment_model_info = (
deployment_model_info: Final = (
{"id": "deployment-1"}
if deployment_declared is None
else {"id": "deployment-1", "supports_mid_conversation_system": deployment_declared}
@ -10896,7 +10899,7 @@ def test_declared_mid_conversation_system_support_prefers_the_deployment_over_th
)
def test_declared_mid_conversation_system_support_is_none_when_the_model_lookup_raises(monkeypatch):
def test_declared_mid_conversation_system_support_is_none_when_the_model_lookup_raises(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setitem(litellm.model_cost, "hosted_vllm/unreadable-probe", "not a model info entry")
litellm.get_model_info.cache_clear()
assert litellm.utils.declared_mid_conversation_system_support("unreadable-probe", "hosted_vllm", None) is None