test: close mutation-testing gaps in container, skills and openai-like config factories

Mutation testing surfaced three factory functions whose tests ran against them
but asserted nothing that a mutation could break, so every planted bug survived.

- litellm/llms/litellm_proxy/skills/code_execution.py: the OpenAI and Anthropic
  tool schemas were unpinned (the Anthropic one was not reached by any test at
  all) and the handler's default fallbacks were unchecked
- litellm/containers/endpoint_factory.py: the endpoints.json contract, the
  generated sync/async function set and the response-type mapping were unpinned
- litellm/llms/openai_like/dynamic_config.py: the generated Responses API config
  class had no coverage of auth header, URL resolution or the store override

The openai_like tests clear _responses_config_cache around each test. Without
that, the module-level cache hands back a class built before the mutation and
the tests pass against mutated code.

Verified by re-running mutmut per scope:
  llms/litellm_proxy  45.2% -> 62.8%  (70 mutants newly killed)
  containers          36.8% -> 84.3%  (45 mutants newly killed)
  llms/openai_like    55.7% -> 66.9%  (34 mutants newly killed)
This commit is contained in:
Yuneng Jiang 2026-08-28 09:28:14 -07:00
parent ca0b951a43
commit 0ec619e7d1
No known key found for this signature in database
3 changed files with 388 additions and 0 deletions

View file

@ -0,0 +1,138 @@
import pytest
from litellm.containers import endpoint_factory
from litellm.containers.endpoint_factory import (
RESPONSE_TYPES,
_load_endpoints_config,
create_sync_endpoint_function,
generate_container_endpoints,
get_all_endpoint_names,
get_async_endpoint_names,
)
from litellm.types.containers.main import (
ContainerFileListResponse,
ContainerFileObject,
DeleteContainerFileResponse,
)
_SYNC_NAMES = [
"list_container_files",
"upload_container_file",
"retrieve_container_file",
"delete_container_file",
"retrieve_container_file_content",
]
_ASYNC_NAMES = ["a" + n for n in _SYNC_NAMES]
class TestEndpointsConfig:
def test_config_exposes_every_declared_endpoint(self):
config = _load_endpoints_config()
assert [e["name"] for e in config["endpoints"]] == _SYNC_NAMES
def test_every_endpoint_declares_the_keys_the_factory_reads(self):
for endpoint in _load_endpoints_config()["endpoints"]:
assert set(endpoint) >= {
"name",
"async_name",
"path",
"method",
"path_params",
"response_type",
}
def test_async_name_is_the_sync_name_prefixed_with_a(self):
for endpoint in _load_endpoints_config()["endpoints"]:
assert endpoint["async_name"] == "a" + endpoint["name"]
def test_config_is_reread_rather_than_shared_between_callers(self):
first = _load_endpoints_config()
first["endpoints"].clear()
assert len(_load_endpoints_config()["endpoints"]) == len(_SYNC_NAMES)
class TestResponseTypeMapping:
def test_mapping_resolves_every_named_response_type(self):
assert RESPONSE_TYPES == {
"ContainerFileListResponse": ContainerFileListResponse,
"ContainerFileObject": ContainerFileObject,
"DeleteContainerFileResponse": DeleteContainerFileResponse,
}
@pytest.mark.parametrize(
"endpoint_name,expected",
[
("list_container_files", ContainerFileListResponse),
("upload_container_file", ContainerFileObject),
("retrieve_container_file", ContainerFileObject),
("delete_container_file", DeleteContainerFileResponse),
],
)
def test_each_endpoint_maps_to_its_declared_response_type(self, endpoint_name, expected):
config = next(e for e in _load_endpoints_config()["endpoints"] if e["name"] == endpoint_name)
assert RESPONSE_TYPES[config["response_type"]] is expected
def test_raw_response_type_is_deliberately_unmapped(self):
config = next(
e for e in _load_endpoints_config()["endpoints"] if e["name"] == "retrieve_container_file_content"
)
assert config["response_type"] == "raw"
assert RESPONSE_TYPES.get(config["response_type"]) is None
class TestGeneratedEndpoints:
def test_generates_exactly_one_sync_and_one_async_function_per_endpoint(self):
assert set(generate_container_endpoints()) == set(_SYNC_NAMES) | set(_ASYNC_NAMES)
def test_every_generated_value_is_callable(self):
assert all(callable(f) for f in generate_container_endpoints().values())
def test_sync_and_async_entries_are_distinct_objects(self):
endpoints = generate_container_endpoints()
for name in _SYNC_NAMES:
assert endpoints[name] is not endpoints["a" + name]
def test_each_call_builds_fresh_functions(self):
assert (
generate_container_endpoints()["list_container_files"]
is not generate_container_endpoints()["list_container_files"]
)
def test_module_exports_are_wired_and_not_none(self):
for name in _SYNC_NAMES + _ASYNC_NAMES:
assert getattr(endpoint_factory, name) is not None
class TestEndpointNameHelpers:
def test_all_endpoint_names_interleaves_sync_then_async_per_endpoint(self):
expected = [n for name in _SYNC_NAMES for n in (name, "a" + name)]
assert get_all_endpoint_names() == expected
def test_async_endpoint_names_are_only_the_async_ones(self):
assert get_async_endpoint_names() == _ASYNC_NAMES
def test_async_names_are_a_strict_subset_of_all_names(self):
assert set(get_async_endpoint_names()) < set(get_all_endpoint_names())
class TestSyncEndpointFactory:
def test_returns_a_callable_for_a_minimal_config(self):
assert callable(
create_sync_endpoint_function({"name": "x", "response_type": "ContainerFileObject", "path_params": []})
)
def test_missing_path_params_defaults_to_empty_rather_than_raising(self):
assert callable(create_sync_endpoint_function({"name": "x", "response_type": "ContainerFileObject"}))
def test_unknown_response_type_is_tolerated_at_build_time(self):
assert callable(
create_sync_endpoint_function({"name": "x", "response_type": "NotARealType", "path_params": []})
)
def test_missing_name_is_a_build_time_error(self):
with pytest.raises(KeyError):
create_sync_endpoint_function({"response_type": "ContainerFileObject"})
def test_missing_response_type_is_a_build_time_error(self):
with pytest.raises(KeyError):
create_sync_endpoint_function({"name": "x"})

View file

@ -0,0 +1,106 @@
import pytest
from litellm.llms.litellm_proxy.skills.code_execution import (
LITELLM_CODE_EXECUTION_TOOL,
CodeExecutionHandler,
LiteLLMInternalTools,
get_litellm_code_execution_tool,
get_litellm_code_execution_tool_anthropic,
)
from litellm.llms.litellm_proxy.skills.constants import (
DEFAULT_MAX_ITERATIONS,
DEFAULT_SANDBOX_TIMEOUT,
)
_DESCRIPTION = (
"Execute Python code in a sandboxed environment. Use this to run code that "
"generates files, processes data, or performs computations. Generated files "
"will be returned directly."
)
class TestInternalToolName:
def test_code_execution_tool_name_is_stable(self):
assert LiteLLMInternalTools.CODE_EXECUTION.value == "litellm_code_execution"
def test_enum_is_str_subclass_so_it_serializes_as_the_bare_name(self):
assert isinstance(LiteLLMInternalTools.CODE_EXECUTION, str)
class TestOpenAIToolSchema:
def test_schema_matches_openai_function_tool_contract_exactly(self):
assert get_litellm_code_execution_tool() == {
"type": "function",
"function": {
"name": "litellm_code_execution",
"description": _DESCRIPTION,
"parameters": {
"type": "object",
"properties": {"code": {"type": "string", "description": "Python code to execute"}},
"required": ["code"],
},
},
}
def test_returns_a_fresh_dict_each_call_so_callers_cannot_mutate_the_shared_one(self):
first = get_litellm_code_execution_tool()
first["function"]["name"] = "clobbered"
assert get_litellm_code_execution_tool()["function"]["name"] == "litellm_code_execution"
def test_singleton_matches_the_factory(self):
assert LITELLM_CODE_EXECUTION_TOOL == get_litellm_code_execution_tool()
class TestAnthropicToolSchema:
def test_schema_matches_anthropic_messages_tool_contract_exactly(self):
assert get_litellm_code_execution_tool_anthropic() == {
"name": "litellm_code_execution",
"description": _DESCRIPTION,
"input_schema": {
"type": "object",
"properties": {"code": {"type": "string", "description": "Python code to execute"}},
"required": ["code"],
},
}
def test_anthropic_shape_is_flat_and_carries_no_openai_only_keys(self):
tool = get_litellm_code_execution_tool_anthropic()
assert "input_schema" in tool
assert "type" not in tool
assert "function" not in tool
assert "parameters" not in tool
def test_returns_a_fresh_dict_each_call(self):
get_litellm_code_execution_tool_anthropic()["name"] = "clobbered"
assert get_litellm_code_execution_tool_anthropic()["name"] == "litellm_code_execution"
def test_both_surfaces_agree_on_name_and_description(self):
openai_tool = get_litellm_code_execution_tool()
anthropic_tool = get_litellm_code_execution_tool_anthropic()
assert anthropic_tool["name"] == openai_tool["function"]["name"]
assert anthropic_tool["description"] == openai_tool["function"]["description"]
assert anthropic_tool["input_schema"] == openai_tool["function"]["parameters"]
class TestHandlerDefaults:
def test_defaults_come_from_constants_when_nothing_is_passed(self):
handler = CodeExecutionHandler()
assert handler.max_iterations == DEFAULT_MAX_ITERATIONS
assert handler.sandbox_timeout == DEFAULT_SANDBOX_TIMEOUT
def test_explicit_values_win_over_the_defaults(self):
handler = CodeExecutionHandler(max_iterations=3, sandbox_timeout=7)
assert handler.max_iterations == 3
assert handler.sandbox_timeout == 7
def test_each_argument_falls_back_independently(self):
assert CodeExecutionHandler(max_iterations=3).sandbox_timeout == DEFAULT_SANDBOX_TIMEOUT
assert CodeExecutionHandler(max_iterations=3).max_iterations == 3
assert CodeExecutionHandler(sandbox_timeout=7).max_iterations == DEFAULT_MAX_ITERATIONS
assert CodeExecutionHandler(sandbox_timeout=7).sandbox_timeout == 7
@pytest.mark.parametrize("falsy", [0, None])
def test_falsy_values_fall_back_to_the_defaults(self, falsy):
handler = CodeExecutionHandler(max_iterations=falsy, sandbox_timeout=falsy)
assert handler.max_iterations == DEFAULT_MAX_ITERATIONS
assert handler.sandbox_timeout == DEFAULT_SANDBOX_TIMEOUT

View file

@ -0,0 +1,144 @@
import pytest
from litellm.llms.openai_like import dynamic_config
from litellm.llms.openai_like.dynamic_config import create_responses_config_class
from litellm.llms.openai_like.json_loader import SimpleProviderConfig
from litellm.types.router import GenericLiteLLMParams
_BASE = {"base_url": "https://api.example.com/v1", "api_key_env": "EXAMPLE_API_KEY"}
def _provider(slug, **overrides):
return SimpleProviderConfig(slug=slug, data={**_BASE, **overrides})
@pytest.fixture(autouse=True)
def _isolate_generated_class_cache():
dynamic_config._responses_config_cache.clear()
yield
dynamic_config._responses_config_cache.clear()
class TestClassCaching:
def test_same_slug_returns_the_identical_class_object(self):
provider = _provider("cache_same_slug")
assert create_responses_config_class(provider) is create_responses_config_class(provider)
def test_cache_is_keyed_on_slug_not_on_the_provider_instance(self):
first = create_responses_config_class(_provider("cache_by_slug"))
second = create_responses_config_class(_provider("cache_by_slug"))
assert first is second
def test_different_slugs_get_different_classes(self):
assert create_responses_config_class(_provider("cache_slug_a")) is not (
create_responses_config_class(_provider("cache_slug_b"))
)
def test_returns_a_class_not_an_instance(self):
assert isinstance(create_responses_config_class(_provider("returns_class")), type)
class TestCustomLlmProvider:
def test_provider_property_reports_the_slug(self):
config = create_responses_config_class(_provider("provider_prop"))()
assert config.custom_llm_provider == "provider_prop"
class TestValidateEnvironment:
def test_explicit_api_key_becomes_a_bearer_header(self):
config = create_responses_config_class(_provider("ve_explicit"))()
headers = config.validate_environment(
headers={}, model="m", litellm_params=GenericLiteLLMParams(api_key="sk-explicit")
)
assert headers["Authorization"] == "Bearer sk-explicit"
def test_api_key_falls_back_to_the_configured_env_var(self, monkeypatch):
monkeypatch.setenv("VE_ENV_KEY", "sk-from-env")
config = create_responses_config_class(_provider("ve_env", api_key_env="VE_ENV_KEY"))()
headers = config.validate_environment(headers={}, model="m", litellm_params=None)
assert headers["Authorization"] == "Bearer sk-from-env"
def test_explicit_key_wins_over_the_env_var(self, monkeypatch):
monkeypatch.setenv("VE_LOSER_KEY", "sk-from-env")
config = create_responses_config_class(_provider("ve_precedence", api_key_env="VE_LOSER_KEY"))()
headers = config.validate_environment(
headers={}, model="m", litellm_params=GenericLiteLLMParams(api_key="sk-wins")
)
assert headers["Authorization"] == "Bearer sk-wins"
def test_no_key_anywhere_leaves_the_header_unset(self, monkeypatch):
monkeypatch.delenv("VE_MISSING_KEY", raising=False)
config = create_responses_config_class(_provider("ve_missing", api_key_env="VE_MISSING_KEY"))()
assert config.validate_environment(headers={}, model="m", litellm_params=None) == {}
def test_existing_headers_are_preserved(self):
config = create_responses_config_class(_provider("ve_preserve"))()
headers = config.validate_environment(
headers={"X-Trace": "abc"},
model="m",
litellm_params=GenericLiteLLMParams(api_key="sk-1"),
)
assert headers["X-Trace"] == "abc"
class TestGetCompleteUrl:
def test_explicit_api_base_gets_the_responses_suffix(self):
config = create_responses_config_class(_provider("url_explicit"))()
assert config.get_complete_url(api_base="https://host/v1", litellm_params={}) == "https://host/v1/responses"
def test_trailing_slash_is_stripped_before_appending(self):
config = create_responses_config_class(_provider("url_slash"))()
assert config.get_complete_url(api_base="https://host/v1/", litellm_params={}) == "https://host/v1/responses"
def test_falls_back_to_the_api_base_env_var(self, monkeypatch):
monkeypatch.setenv("URL_BASE_ENV", "https://from-env/v1")
config = create_responses_config_class(_provider("url_env", api_base_env="URL_BASE_ENV"))()
assert config.get_complete_url(api_base=None, litellm_params={}) == "https://from-env/v1/responses"
def test_falls_back_to_the_configured_base_url_last(self, monkeypatch):
monkeypatch.delenv("URL_UNSET_ENV", raising=False)
config = create_responses_config_class(_provider("url_base_url", api_base_env="URL_UNSET_ENV"))()
assert config.get_complete_url(api_base=None, litellm_params={}) == "https://api.example.com/v1/responses"
def test_explicit_api_base_wins_over_the_env_var(self, monkeypatch):
monkeypatch.setenv("URL_LOSER_ENV", "https://from-env/v1")
config = create_responses_config_class(_provider("url_precedence", api_base_env="URL_LOSER_ENV"))()
assert (
config.get_complete_url(api_base="https://explicit/v1", litellm_params={})
== "https://explicit/v1/responses"
)
def test_no_base_anywhere_raises_naming_the_provider(self):
provider = _provider("url_none")
provider.base_url = None
config = create_responses_config_class(provider)()
with pytest.raises(ValueError, match="url_none"):
config.get_complete_url(api_base=None, litellm_params={})
class TestForceStoreFalse:
def test_force_store_false_overrides_the_caller(self):
config = create_responses_config_class(
_provider("store_forced", special_handling={"force_store_false": True})
)()
params = {"store": True}
config.transform_responses_api_request(
model="m",
input="hi",
response_api_optional_request_params=params,
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert params["store"] is False
def test_without_the_flag_the_callers_store_value_is_left_alone(self):
config = create_responses_config_class(_provider("store_untouched"))()
params = {"store": True}
config.transform_responses_api_request(
model="m",
input="hi",
response_api_optional_request_params=params,
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert params["store"] is True