scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-14 04:09:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 97

test(batches): pin that a pre-call ownership denial stops every batch create branch

Browse source 
LIT-4797 noted an endpoint-level check missed the load-balanced branch
because it dispatches first. The ownership check lives in the pre-call
hook, which runs before branch selection, so a 403 there must keep both
the load-balanced and unified-file branches from reaching the router
This commit is contained in:
Mrinal Chanshetty 2026-10-10 12:53:07 -07:00
parent f673922070
commit e4e1eb9939
1 changed files with 29 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
tests/unit/proxy/batches_endpoints/test_endpoints.py
View file

@ -57,7 +57,7 @@ from litellm.types.llms.openai import BatchJobStatus
from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo
from litellm.types.utils import CredentialItem, LiteLLMBatch, SpecialEnums
from fastapi import Request, Response
from fastapi import HTTPException, Request, Response
# --------------------------------------------------------------------------- #
# Fixtures: distinguishable credentials per model so a wrong/hardcoded model_id
@ -1164,6 +1164,34 @@ async def test_create__uses_acreate_batch_route_type(harness, openai_env_creds):
assert harness.pre_call.call_args.kwargs["route_type"] == "acreate_batch"
@pytest.mark.asyncio
@pytest.mark.parametrize("loadbalancing", [True, False], ids=["load_balanced", "unified_branch"])
async def test_create__pre_call_ownership_denial_stops_every_dispatch_branch(harness, loadbalancing: bool):
set_body(
harness,
{
"input_file_id": "litellm_proxy_unified_id",
"model": "vertex-model",
"endpoint": "/v1/chat/completions",
"completion_window": "24h",
},
)
harness.is_known_model.return_value = True
harness.pre_call.side_effect = HTTPException(status_code=403, detail="does not have access to the file")
with (
patch.object(litellm, "enable_loadbalancing_on_batch_endpoints", loadbalancing),
patch.object(endpoints, "is_base64_encoded_unified_file_id", return_value="unified-xyz"),
patch.object(endpoints, "get_models_from_unified_file_id", return_value=["vertex-model"]),
pytest.raises(ProxyException) as exc,
):
await call_create(harness)
assert exc.value.code == "403", exc.value
harness.router_acreate.assert_not_called()
harness.litellm_acreate.assert_not_called()
def install_managed_files_hook(harness: Harness) -> AsyncMock:
prisma_client = AsyncMock()
managed_files = PROXY_LiteLLMManagedFiles(MagicMock(async_set_cache=AsyncMock()), prisma_client=prisma_client)