scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-11 22:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 124

Merge remote-tracking branch 'origin/litellm_internal_staging' into litellm_/elated-noyce-6fc150

Browse source
This commit is contained in:
Yuneng Jiang 2026-07-06 14:16:02 -07:00
commit 82a5b37030
No known key found for this signature in database
17 changed files with 565 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/litellm_core_utils/exception_mapping_utils.py
View file

@ -2173,7 +2173,7 @@ def exception_type( # type: ignore
litellm_response_headers = _get_response_headers(original_exception=original_exception)
try:
error_str = redact_string(str(original_exception)) if _ENABLE_SECRET_REDACTION else str(original_exception)
if model:
if model or custom_llm_provider:
if hasattr(original_exception, "message"):
error_str = (
redact_string(str(original_exception.message))

33
litellm/llms/azure/responses/transformation.py
View file

@ -205,7 +205,7 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
#########################################################
########## DELETE RESPONSE API TRANSFORMATION ##############
#########################################################
def _construct_url_for_response_id_in_path(self, api_base: str, response_id: str) -> str:
def _construct_url_for_response_id_in_path(self, api_base: str, response_id: str, path_suffix: str = "") -> str:
"""
Constructs a URL for the API request with the response_id in the path.
"""
@ -218,14 +218,14 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
# Remove trailing slash if present to avoid double slashes
path = parsed_url.path.rstrip("/")
encoded_response_id = encode_url_path_segment(response_id, field_name="response_id")
new_path = f"{path}/{encoded_response_id}"
new_path = f"{path}/{encoded_response_id}{path_suffix}"
# Reconstruct the URL with all original components but with the modified path
constructed_url = urlunparse(
(
parsed_url.scheme, # http, https
parsed_url.netloc, # domain name, port
new_path, # path with response_id added
new_path,
parsed_url.params, # parameters
parsed_url.query, # query string
parsed_url.fragment, # fragment
@ -288,7 +288,9 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
limit: int = 20,
order: Literal["asc", "desc"] = "desc",
) -> Tuple[str, Dict]:
url = self._construct_url_for_response_id_in_path(api_base=api_base, response_id=response_id) + "/input_items"
url = self._construct_url_for_response_id_in_path(
api_base=api_base, response_id=response_id, path_suffix="/input_items"
)
params: Dict[str, Any] = {}
if after is not None:
params["after"] = after
@ -322,27 +324,8 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
This function handles URLs with query parameters by inserting the response_id
at the correct location (before any query parameters).
"""
from urllib.parse import urlparse, urlunparse
# Parse the URL to separate its components
parsed_url = urlparse(api_base)
# Insert the response_id and /cancel at the end of the path component
# Remove trailing slash if present to avoid double slashes
path = parsed_url.path.rstrip("/")
encoded_response_id = encode_url_path_segment(response_id, field_name="response_id")
new_path = f"{path}/{encoded_response_id}/cancel"
# Reconstruct the URL with all original components but with the modified path
cancel_url = urlunparse(
(
parsed_url.scheme, # http, https
parsed_url.netloc, # domain name, port
new_path, # path with response_id and /cancel added
parsed_url.params, # parameters
parsed_url.query, # query string
parsed_url.fragment, # fragment
)
cancel_url = self._construct_url_for_response_id_in_path(
api_base=api_base, response_id=response_id, path_suffix="/cancel"
)
data: Dict = {}

3
litellm/proxy/hooks/responses_id_security.py
View file

@ -44,6 +44,7 @@ class ResponsesIDSecurity(CustomLogger):
"aget_responses",
"adelete_responses",
"acancel_responses",
"alist_input_items",
}
if call_type not in responses_api_call_types:
return None
@ -54,7 +55,7 @@ class ResponsesIDSecurity(CustomLogger):
original_response_id, user_id, team_id = self._decrypt_response_id(previous_response_id)
self.check_user_access_to_response_id(user_id, team_id, user_api_key_dict)
data["previous_response_id"] = original_response_id
elif call_type in {"aget_responses", "adelete_responses", "acancel_responses"}:
elif call_type in {"aget_responses", "adelete_responses", "acancel_responses", "alist_input_items"}:
response_id = data.get("response_id")
if response_id and self._is_encrypted_response_id(response_id):

40
tests/e2e/e2e_gateway.py
View file

@ -9,6 +9,7 @@ Gateway's key/customer methods for cleanup. Read-backs are eventually consistent
from __future__ import annotations
import time
import warnings
from collections.abc import Callable
from dataclasses import dataclass
@ -18,6 +19,7 @@ from e2e_http import (
Result,
StreamingResponse,
Success,
is_ok,
unwrap,
)
from models import (
@ -32,8 +34,14 @@ from models import (
KeyInfo,
KeyInfoParams,
KeyInfoResponse,
LiteLLMParamsBody,
ModelDeleteBody,
ModelInfoBody,
ModelInfoEntry,
ModelInfoResponse,
ModelMode,
ModelNewBody,
ModelNewResponse,
OcrBody,
OcrResponse,
SpendLogRow,
@ -111,6 +119,38 @@ class Gateway:
)
).data
def create_model(
self,
model_name: str,
litellm_params: LiteLLMParamsBody,
mode: ModelMode | None = None,
) -> str:
"""Register a deployment under `model_name` (id == model_name) and return the
model_id. add_deployment runs synchronously in /model/new, so the model is
callable as soon as this returns."""
return unwrap(
self.transport.post(
"/model/new",
headers=self.transport.master,
json=ModelNewBody(
model_name=model_name,
litellm_params=litellm_params,
model_info=ModelInfoBody(id=model_name, mode=mode),
),
response_type=ModelNewResponse,
)
).model_id
def delete_model(self, model_id: str) -> None:
result = self.transport.post(
"/model/delete",
headers=self.transport.master,
json=ModelDeleteBody(id=model_id),
response_type=NoBody,
)
if not is_ok(result):
warnings.warn(f"delete_model({model_id!r}) failed: {result}", stacklevel=2)
# ---- LLM calls ------------------------------------------------------
def chat(self, key: str, body: ChatBody) -> Result[ChatResponse]:

37
tests/e2e/llm_translation/endpoints_client.py
View file

@ -14,15 +14,8 @@ from dataclasses import dataclass
from pydantic import BaseModel
from e2e_gateway import Gateway, build_gateway
from e2e_http import NoBody, StreamingResponse, is_ok, unwrap
from models import (
ChatMessage,
LiteLLMParamsBody,
ModelDeleteBody,
ModelInfoBody,
ModelNewBody,
ModelNewResponse,
)
from e2e_http import StreamingResponse
from models import ChatMessage, LiteLLMParamsBody
class ResponsesRequest(BaseModel):
@ -136,32 +129,10 @@ class EndpointsClient:
gateway: Gateway
def create_model(self, model_name: str, litellm_params: LiteLLMParamsBody) -> str:
"""Register a deployment under `model_name` (id == model_name) and return the
model_id. add_deployment runs synchronously in /model/new, so the model is
callable as soon as this returns."""
return unwrap(
self.gateway.transport.post(
"/model/new",
headers=self.gateway.transport.master,
json=ModelNewBody(
model_name=model_name,
litellm_params=litellm_params,
model_info=ModelInfoBody(id=model_name),
),
response_type=ModelNewResponse,
)
).model_id
return self.gateway.create_model(model_name, litellm_params)
def delete_model(self, model_id: str) -> None:
result = self.gateway.transport.post(
"/model/delete",
headers=self.gateway.transport.master,
json=ModelDeleteBody(id=model_id),
response_type=NoBody,
)
if not is_ok(result):
import warnings
warnings.warn(f"delete_model({model_id!r}) failed: {result}", stacklevel=2)
self.gateway.delete_model(model_id)
def _send(self, path: str, key: str, body: BaseModel) -> StreamingResponse:
return self.gateway.transport.send(

24
tests/e2e/models.py
View file

@ -216,6 +216,25 @@ class SpendLogsParams(BaseModel):
api_key: str | None = None
class SpendLogsPageParams(BaseModel):
"""Query for /spend/logs/v2, which requires an explicit date window and
serves pages of at most 100 rows."""
start_date: str
end_date: str
page: int
page_size: int
api_key: str | None = None
class SpendLogsPage(BaseModel):
data: list[SpendLogRow] = []
total: int
page: int
page_size: int
total_pages: int
# ---------- spend calculate ----------
@ -367,9 +386,12 @@ class LiteLLMParamsBody(BaseModel):
output_cost_per_token: float | None = None
ModelMode = Literal["batch", "realtime", "image_generation"]
class ModelInfoBody(BaseModel):
id: str
mode: Literal["batch", "realtime", "image_generation"] | None = None
mode: ModelMode | None = None
class ModelNewBody(BaseModel):

4
tests/e2e/spend_tracking/SPEND_TRACKING_COVERAGE_MATRIX.md
View file

@ -43,6 +43,7 @@ proxy + SpendLogs rows. Status: `covered` / `partial` / `gap`.
| Tag | `test_update_daily_tag_spend.py` | partial | yes (`test_tag_spend_matches_sum_of_tagged_logs`) |
| End-user | `test_proxy_update_spend.py` | covered | yes |
| Spend == sum(logs) consistency | none | gap | yes (key + tag aggregate == sum of rows) |
| Concurrent increments (one key, parallel writers) | `tests/spend_tracking_tests/test_spend_accuracy_tests.py` (burst) | partial | yes (`test_burst_of_concurrent_calls_loses_no_spend`) |
## Spend read endpoints (verification surface)
@ -51,6 +52,7 @@ proxy + SpendLogs rows. Status: `covered` / `partial` / `gap`.
| `/spend/logs` (request_id / api_key) | `test_spend_management_endpoints.py` | covered | yes (primary read path; `test_spend_logs_endpoint_returns_spend` asserts 200 + spend, never 5xx) |
| `/spend/calculate` | `local_testing/test_spend_calculate_endpoint.py` | covered | yes (`test_spend_calculate_returns_nonzero_cost`) |
| `/spend/tags` | `test_spend_management_endpoints.py` | partial | yes (tag accuracy test) |
| `/spend/logs/v2` pagination (total/total_pages/out-of-range) | `test_spend_query_optimization.py` | covered | yes (`test_spend_logs_v2_pagination_caps_pages_and_keeps_total`; filter takes the hashed token, not the raw key) |
| whole spend GET surface (22 routes) | unit per-handler | partial | yes (`test_spend_routes.py` probes each for 404/5xx) |
## What this suite pins
@ -69,6 +71,8 @@ proxy + SpendLogs rows. Status: `covered` / `partial` / `gap`.
| `test_failure_call_writes_failure_status_row` | failed call -> `status=failure`, `spend=0` |
| `test_spend_calculate_returns_nonzero_cost` | cost-map smoke (no batch wait) |
| `test_spend_logs_endpoint_returns_spend` | `/spend/logs` returns 200 + the key's spend, never a 5xx (intermittent-500 regression) |
| `test_burst_of_concurrent_calls_loses_no_spend` | N parallel calls on one key: N distinct costed rows, key aggregate == sum (no lost increments) |
| `test_spend_logs_v2_pagination_caps_pages_and_keeps_total` | `/spend/logs/v2` page cap, stable total on out-of-range page, zero total on no-match filter |
| `test_spend_routes.py` (23) | no spend route 404s or 5xxs |
## Design + timing

41
tests/e2e/spend_tracking/conftest.py
View file

@ -1,16 +1,55 @@
"""Spend-tracking suite's `client` fixture.
"""Spend-tracking suite's `client` fixture and driver-model registration.
The shared lifecycle (resources/scoped_key), proxy liveness skip, and e2e marker
live in the parent tests/e2e/conftest.py. SpendClient exposes the shared Gateway
(GatewayProvider), so the `resources` fixture cleans up keys and customers this
suite creates.
The suite drives real calls through three deployments. On the stage gateway they
are baked into the proxy config; on a local dev proxy they usually are not, so
`driver_models` registers whichever are missing via /model/new and deletes only
the ones it created, never a config-baked deployment. Each registration carries
the provider key from the test runner's env when set (so a local proxy whose
container env lacks the key still works); otherwise it falls back to an
os.environ reference resolved from the proxy's own env, the stage convention.
"""
import os
from typing import Iterator
import pytest
from models import LiteLLMParamsBody
from spend_e2e_client import SpendClient, build_client
def _driver_params(provider_model: str, env_var: str) -> LiteLLMParamsBody:
return LiteLLMParamsBody(
model=provider_model,
api_key=os.environ.get(env_var) or f"os.environ/{env_var}",
)
DRIVER_MODELS: tuple[tuple[str, str, str], ...] = (
("gemini-2.5-flash", "gemini/gemini-2.5-flash", "GEMINI_API_KEY"),
("claude-haiku-4-5", "anthropic/claude-haiku-4-5", "ANTHROPIC_API_KEY"),
("openai-text-embedding-3-small", "openai/text-embedding-3-small", "OPENAI_API_KEY"),
)
@pytest.fixture(scope="session")
def client() -> SpendClient:
return build_client()
@pytest.fixture(scope="session", autouse=True)
def driver_models(client: SpendClient) -> Iterator[None]:
existing = frozenset(entry.model_name for entry in client.gateway.model_info())
created = tuple(
client.gateway.create_model(name, _driver_params(provider_model, env_var))
for name, provider_model, env_var in DRIVER_MODELS
if name not in existing
)
yield
for model_id in created:
client.gateway.delete_model(model_id)

25
tests/e2e/spend_tracking/spend_e2e_client.py
View file

@ -15,6 +15,7 @@ import os
import time
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from e2e_config import unique_marker
from e2e_http import (
@ -39,6 +40,8 @@ from models import (
SpendCalculateBody,
SpendCalculateResponse,
SpendLogRow,
SpendLogsPage,
SpendLogsPageParams,
SpendTagsResponse,
TagSpend,
)
@ -180,6 +183,28 @@ class SpendClient:
time.sleep(self.gateway.poll_interval)
return spend
def spend_logs_page(
self, *, api_key: str | None, page: int, page_size: int
) -> SpendLogsPage:
"""One page of /spend/logs/v2 over a window wide enough to contain every
row this test run wrote (the endpoint requires explicit dates)."""
now = datetime.now(timezone.utc)
fmt = "%Y-%m-%d %H:%M:%S"
return unwrap(
self.gateway.transport.get(
"/spend/logs/v2",
headers=self.gateway.transport.master,
params=SpendLogsPageParams(
start_date=(now - timedelta(days=1)).strftime(fmt),
end_date=(now + timedelta(days=1)).strftime(fmt),
page=page,
page_size=page_size,
api_key=api_key,
),
response_type=SpendLogsPage,
)
)
def probe(self, path: str, *, params: DateRangeParams) -> ProbeResult:
return self.gateway.transport.probe(path, params=params)

19
tests/e2e/spend_tracking/test_spend_routes.py
View file

@ -27,13 +27,21 @@ pytestmark = pytest.mark.e2e
# Verified present and responsive on a live proxy. One per row of the spend
# surface: key / user / team / org / customer aggregation, model-cost, tags,
# activity.
# activity. Most are include_in_schema=False, so keep this list exhaustive by
# hand; the schema test below only auto-catches the visible minority. Excluded
# by design: path-param routes (/spend/logs/ui/{request_id}), POST readers
# (/spend/calculate has its own test, /global/spend/end_users), mutating
# POSTs (/global/spend/reset, /global/spend/refresh), and /provider/budgets,
# which 500s whenever router_settings.provider_budget_config is absent, so it
# is only probeable on a proxy configured with provider budget routing.
SPEND_ROUTES = (
"/spend/keys",
"/spend/users",
"/spend/tags",
"/spend/logs",
"/spend/logs/ui",
"/spend/logs/v2",
"/spend/logs/session/ui",
"/global/spend",
"/global/spend/keys",
"/global/spend/teams",
@ -43,9 +51,18 @@ SPEND_ROUTES = (
"/global/spend/tags",
"/global/spend/logs",
"/global/spend/all_tag_names",
"/global/all_end_users",
"/global/activity",
"/global/activity/model",
"/global/activity/exceptions",
"/global/activity/exceptions/deployment",
"/user/daily/activity",
"/user/daily/activity/aggregated",
"/team/daily/activity",
"/organization/daily/activity",
"/customer/daily/activity",
"/end_user/daily/activity",
"/tag/daily/activity",
"/key/list",
"/user/list",
"/team/list",

103
tests/e2e/spend_tracking/test_spend_tracking_e2e.py
View file

@ -17,12 +17,13 @@ fails the test; a pricing or token-count drift does not.
import time
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
import pytest
from e2e_http import Success
from e2e_http import Result, Success
from lifecycle import ResourceManager
from models import SpendLogs, SpendLogsParams
from models import ChatResponse, SpendLogs, SpendLogsParams
from spend_e2e_client import SpendClient, SpendLogRow, is_ok, unique_marker, unwrap
pytestmark = pytest.mark.e2e
@ -202,6 +203,104 @@ def test_key_spend_equals_sum_of_logs(client: SpendClient, scoped_key: str) -> N
), f"key aggregate {key_spend} != sum of logs {logs_total}; rows: {_summarize(rows)}"
def test_burst_of_concurrent_calls_loses_no_spend(
client: SpendClient, scoped_key: str
) -> None:
"""Six concurrent calls on one key: every call lands its own spend row under a
distinct request_id and the key aggregate equals the sum of the rows.
Sequential accuracy is covered by test_key_spend_equals_sum_of_logs; this pins
the concurrent increment path (parallel writers racing on one key's counter),
where a lost update can never be reproduced by sequential calls."""
burst = 6
def call(idx: int) -> Result[ChatResponse]:
return client.chat(
scoped_key,
"gemini-2.5-flash",
f"burst call {idx} {unique_marker()}",
max_tokens=16,
)
with ThreadPoolExecutor(max_workers=burst) as pool:
results = tuple(pool.map(call, range(burst)))
failed = [r for r in results if not is_ok(r)]
assert not failed, f"{len(failed)}/{burst} burst calls failed; first: {failed[0]}"
rows = client.poll_logs_for_key(
scoped_key,
min_rows=burst,
predicate=lambda rs: len([r for r in rs if (r.spend or 0) > 0]) >= burst,
)
costed = [r for r in rows if (r.spend or 0) > 0]
assert len(costed) >= burst, (
f"only {len(costed)}/{burst} burst calls produced a costed row - "
f"rows lost under concurrency: {_summarize(rows)}"
)
request_ids = [r.request_id for r in costed]
assert len(set(request_ids)) == len(request_ids), (
f"concurrent rows collapsed onto shared request_ids: {_summarize(rows)}"
)
logs_total = sum((r.spend or 0) for r in rows)
key_spend = client.poll_key_spend(scoped_key, minimum=logs_total * 0.999)
assert _approx_equal(key_spend, logs_total), (
f"key aggregate {key_spend} != sum of {len(rows)} rows {logs_total} - "
f"spend increments lost under concurrency: {_summarize(rows)}"
)
def test_spend_logs_v2_pagination_caps_pages_and_keeps_total(
client: SpendClient, scoped_key: str
) -> None:
"""/spend/logs/v2 pagination contract for the key filter: page_size caps the
rows returned, total counts every row for the filter (so with page_size=1,
total_pages == total), a page past the end returns no rows while reporting
the same total (an out-of-range page must not reset the count the UI
paginates by), and a filter matching nothing reports zero without erroring.
Unlike /spend/logs, the v2 filter matches the hashed token exactly as stored
on the row (the form the UI passes), not the raw sk- key, so the filter value
is read off the rows the poll returned."""
for _ in range(2):
_ = unwrap(
client.chat(
scoped_key,
"gemini-2.5-flash",
f"page fodder {unique_marker()}",
max_tokens=16,
)
)
rows = client.poll_logs_for_key(
scoped_key, min_rows=2, predicate=lambda rs: sum((r.spend or 0) for r in rs) > 0
)
hashed_key = rows[0].api_key
assert hashed_key, f"polled rows carry no api_key: {_summarize(rows)}"
first = client.spend_logs_page(api_key=hashed_key, page=1, page_size=1)
assert first.total >= 2, f"expected >=2 rows for the key, got total={first.total}"
assert len(first.data) == 1, f"page_size=1 returned {len(first.data)} rows"
assert first.total_pages == first.total, (
f"page_size=1 must give one page per row: "
f"total={first.total} total_pages={first.total_pages}"
)
beyond = client.spend_logs_page(
api_key=hashed_key, page=first.total_pages + 7, page_size=1
)
assert beyond.data == [], f"out-of-range page returned rows: {beyond.data}"
assert beyond.total == first.total, (
f"out-of-range page changed the total: {beyond.total} != {first.total}"
)
nomatch = client.spend_logs_page(
api_key=f"sk-no-such-key-{unique_marker()}", page=1, page_size=1
)
assert nomatch.total == 0 and nomatch.data == [], (
f"filter matching nothing must report zero: "
f"total={nomatch.total} rows={len(nomatch.data)}"
)
def test_request_tags_round_trip(client: SpendClient, scoped_key: str) -> None:
tag = f"e2e-spend-{unique_marker()}"
_ = unwrap(

148
tests/e2e/test_e2e_gateway.py Normal file
View file

@ -0,0 +1,148 @@
"""Unit coverage for the Gateway model-management surface (create_model /
delete_model).
The batches conftest and several llm_translation tests register deployments at
runtime through gateway.create_model; when that method went missing, every batch
test errored at fixture setup (AttributeError) before a single request reached
the proxy. This pins the surface with a typed fake Transport so a rename or
signature drift fails here instead of in a live stage run.
"""
from dataclasses import dataclass, field
from pydantic import BaseModel
from batches.batch_client import BatchClient
from e2e_gateway import Gateway
from e2e_http import (
AuthHeaders,
FileUploadForm,
ProbeResult,
Result,
StreamingResponse,
Success,
)
from models import (
LiteLLMParamsBody,
ModelDeleteBody,
ModelNewBody,
ModelNewResponse,
)
@dataclass
class _RecordingTransport:
"""Typed fake fulfilling the Transport protocol; records every post and
answers with a canned success so the test asserts on what was sent."""
posts: list[tuple[str, BaseModel]] = field(default_factory=list)
def post[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
self.posts.append((path, json))
payload = (
{"model_id": "registered-id"} if response_type is ModelNewResponse else {}
)
return Success(data=response_type.model_validate(payload))
def stream(
self, path: str, *, headers: BaseModel, json: BaseModel
) -> StreamingResponse:
raise AssertionError("stream is not part of model management")
def send(
self,
path: str,
*,
headers: BaseModel,
json: BaseModel,
params: BaseModel | None = None,
stream: bool = False,
) -> StreamingResponse:
raise AssertionError("send is not part of model management")
def get[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
params: BaseModel,
response_type: type[R],
) -> Result[R]:
raise AssertionError("get is not part of model management")
def delete[R: BaseModel](
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
) -> Result[R]:
raise AssertionError("delete is not part of model management")
def probe(self, path: str, *, params: BaseModel) -> ProbeResult:
raise AssertionError("probe is not part of model management")
def upload[R: BaseModel](
self,
path: str,
*,
headers: BaseModel,
form: FileUploadForm,
filename: str,
content: bytes,
params: BaseModel | None = None,
response_type: type[R],
) -> Result[R]:
raise AssertionError("upload is not part of model management")
def download(self, path: str, *, headers: BaseModel) -> StreamingResponse:
raise AssertionError("download is not part of model management")
def bearer(self, key: str) -> AuthHeaders:
return AuthHeaders(authorization=f"Bearer {key}")
@property
def master(self) -> AuthHeaders:
return self.bearer("sk-test-master")
def test_gateway_create_model_registers_deployment_and_returns_model_id() -> None:
transport = _RecordingTransport()
gateway = Gateway(transport=transport)
model_id = gateway.create_model(
"e2e-test-model", LiteLLMParamsBody(model="openai/gpt-4o-mini")
)
assert model_id == "registered-id"
path, body = transport.posts[0]
assert path == "/model/new"
assert isinstance(body, ModelNewBody)
assert body.model_name == "e2e-test-model"
assert body.model_info.id == "e2e-test-model"
assert body.model_info.mode is None
def test_batch_client_create_model_registers_a_batch_mode_deployment() -> None:
transport = _RecordingTransport()
client = BatchClient(gateway=Gateway(transport=transport))
model_id = client.create_model(
"e2e-batch-model", LiteLLMParamsBody(model="openai/gpt-4o-mini")
)
assert model_id == "registered-id"
path, body = transport.posts[0]
assert path == "/model/new"
assert isinstance(body, ModelNewBody)
assert body.model_info.mode == "batch"
def test_gateway_delete_model_posts_the_model_id() -> None:
transport = _RecordingTransport()
gateway = Gateway(transport=transport)
gateway.delete_model("registered-id")
path, body = transport.posts[0]
assert path == "/model/delete"
assert isinstance(body, ModelDeleteBody)
assert body.id == "registered-id"

52
tests/e2e/test_transport.py Normal file
View file

@ -0,0 +1,52 @@
"""Unit coverage for SplitTransport path routing (is_control_plane_path).
Model-management calls (/model/new, /model/delete, /model/info) must go to the
control plane: the data-plane gateway does not serve management routes, so a
misrouted /model/new 404s and takes down every suite that registers deployments
at runtime (llm_translation, batches, access_control). /models must stay on the
data plane; it is the OpenAI-compatible list-models route, not a management
route.
"""
import pytest
from transport import is_control_plane_path
@pytest.mark.parametrize(
"path",
[
"/model/new",
"/model/delete",
"/model/update",
"/model/info",
"/key/generate",
"/budget/new",
"/spend/logs",
"/end_user/daily/activity",
"/user/daily/activity",
"/team/daily/activity",
"/tag/daily/activity",
],
)
def test_management_routes_go_to_the_control_plane(path: str) -> None:
assert is_control_plane_path(path), (
f"{path} is a management route; sending it to the data plane 404s"
)
@pytest.mark.parametrize(
"path",
[
"/models",
"/v1/models",
"/chat/completions",
"/v1/messages",
"/embeddings",
"/anthropic/v1/messages",
],
)
def test_llm_routes_stay_on_the_data_plane(path: str) -> None:
assert not is_control_plane_path(path), (
f"{path} is an LLM route; it must go to the data plane"
)

3
tests/e2e/transport.py
View file

@ -202,9 +202,10 @@ CONTROL_PLANE_PREFIXES: tuple[str, ...] = (
"/team",
"/organization",
"/customer",
"/end_user",
"/tag",
"/budget",
"/model/info",
"/model/",
"/spend",
"/global",
"/openapi.json",

23
tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py
View file

@ -626,3 +626,26 @@ def test_replicate_422_maps_to_unprocessable_entity():
)
assert excinfo.value.llm_provider == "replicate"
def test_upstream_4xx_without_model_maps_to_bad_request():
"""Responses API follow-ups (cancel/get/delete) call ``exception_type`` with
``model=None``; the provider mapping used to be gated on ``if model:``, so an
upstream 400 like Azure's "Cannot cancel a synchronous response." fell through to
the generic 500 APIConnectionError instead of surfacing as a 400."""
from litellm.llms.base_llm.chat.transformation import BaseLLMException
original_exception = BaseLLMException(
status_code=400,
message='{"error": {"message": "Cannot cancel a synchronous response.", "type": "invalid_request_error"}}',
)
with pytest.raises(litellm.BadRequestError) as excinfo:
exception_type(
model=None,
original_exception=original_exception,
custom_llm_provider="azure",
)
assert excinfo.value.status_code == 400
assert "Cannot cancel a synchronous response." in excinfo.value.message

18
tests/test_litellm/llms/azure/response/test_azure_transformation.py
View file

@ -337,6 +337,24 @@ class TestAzureResponsesAPIConfig:
assert url == expected_url
assert data == {}
def test_azure_list_input_items_request_url_path_before_query(self):
from litellm.types.router import GenericLiteLLMParams
api_base = "https://test.openai.azure.com/openai/responses?api-version=2025-03-01-preview"
url, params = self.config.transform_list_input_items_request(
response_id="resp_test123",
api_base=api_base,
litellm_params=GenericLiteLLMParams(api_version="2025-03-01-preview"),
headers={},
)
assert (
url
== "https://test.openai.azure.com/openai/responses/resp_test123/input_items?api-version=2025-03-01-preview"
)
assert params == {"limit": 20, "order": "desc"}
def test_azure_cancel_response_api_response(self):
"""Test Azure cancel response API response transformation"""
from unittest.mock import Mock

56
tests/test_litellm/test_responses_id_security.py
View file

@ -519,6 +519,62 @@ class TestAsyncPreCallHook:
assert "team" in exc_info.value.detail.lower()
@pytest.mark.asyncio
async def test_async_pre_call_hook_alist_input_items_decrypts_response_id(
self, responses_id_security, mock_user_api_key_dict, mock_cache
):
data = {"response_id": "resp_encrypted_789"}
with patch.object(
responses_id_security, "_is_encrypted_response_id", return_value=True
):
with patch.object(
responses_id_security,
"_decrypt_response_id",
return_value=("resp_original_789", "test-user-123", "test-team-123"),
):
result = await responses_id_security.async_pre_call_hook(
user_api_key_dict=mock_user_api_key_dict,
cache=mock_cache,
data=data,
call_type="alist_input_items",
)
assert result is not None
assert result["response_id"] == "resp_original_789"
@pytest.mark.asyncio
async def test_async_pre_call_hook_alist_input_items_team_security(
self, responses_id_security, mock_cache
):
mock_auth_team_a = MagicMock()
mock_auth_team_a.user_id = None
mock_auth_team_a.team_id = "team-a"
mock_auth_team_a.user_role = None
data = {"response_id": "resp_encrypted_team_b"}
with patch.object(
responses_id_security, "_is_encrypted_response_id", return_value=True
):
with patch.object(
responses_id_security,
"_decrypt_response_id",
return_value=("resp_original_team_b", None, "team-b"),
):
with patch("litellm.proxy.proxy_server.general_settings", {}):
with pytest.raises(HTTPException) as exc_info:
await responses_id_security.async_pre_call_hook(
user_api_key_dict=mock_auth_team_a,
cache=mock_cache,
data=data,
call_type="alist_input_items",
)
assert exc_info.value.status_code == 403
assert "team" in exc_info.value.detail.lower()
class TestAsyncPostCallSuccessHook:
"""Test async_post_call_success_hook function"""