test(batches): restore alias-resolver coverage and fix harness after staging merge

Merging staging surfaced two test gaps. Restoring test_router.py to staging's
version while re-applying the new tests accidentally dropped the original
get_deployment_model_for_alias cases, which router_code_coverage.py (AST
call-graph) needs to see the public method exercised by a router-named test.

Staging also added an exact-kwargs batch-create contract test whose router is a
spec'd MagicMock. The create path now resolves the request alias to the
deployment's real model via get_deployment_model_for_alias, so the unconfigured
mock returned a MagicMock for "model". Wire that seam to the CREDS model so the
swap mirrors production and the payload assertion stays meaningful.
This commit is contained in:
Kent 2026-06-29 19:15:14 +08:00
parent 644f7fec86
commit f1ac51d448
2 changed files with 50 additions and 0 deletions

View file

@ -153,6 +153,13 @@ def _creds_lookup(*, model_id: str) -> Dict[str, str]:
return dict(CREDS[model_id])
def _alias_lookup(*, model_id: str) -> str:
# The endpoint swaps the request model for the deployment's real provider
# model before calling the provider; mirror that with the CREDS model so a
# wrong/hardcoded model_id KeyErrors instead of hiding.
return CREDS[model_id]["model"]
@pytest.fixture
def harness():
"""Seam harness. Patches only true I/O boundaries; pure encode/decode/merge
@ -169,6 +176,7 @@ def harness():
router.get_deployment_credentials_with_provider = MagicMock(
side_effect=_creds_lookup
)
router.get_deployment_model_for_alias = MagicMock(side_effect=_alias_lookup)
read_body = AsyncMock(side_effect=lambda request: body_holder["body"])
pre_call = AsyncMock(side_effect=lambda **kw: (body_holder["body"], MagicMock()))

View file

@ -5174,6 +5174,48 @@ def test_is_deployment_blocked_static_helper_reflects_blocked_flag():
)
def test_get_deployment_model_for_alias_resolves_underlying_model():
"""
The proxy batch-create path resolves a model-group alias to its deployment
so it can hand the provider the deployment's real model id, not the alias.
get_llm_provider cannot resolve a proxy alias, so without this the Bedrock
batch transform receives the alias as a modelId and AWS rejects it.
"""
router = litellm.Router(
model_list=[
{
"model_name": "bedrock-batch-haiku",
"litellm_params": {
"model": "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0",
"aws_region_name": "us-east-1",
},
"model_info": {"id": "bedrock-batch-dep-0"},
}
]
)
assert (
router.get_deployment_model_for_alias(model_id="bedrock-batch-haiku")
== "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
)
# Resolving by deployment id returns the same underlying model.
assert (
router.get_deployment_model_for_alias(model_id="bedrock-batch-dep-0")
== "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
)
def test_get_deployment_model_for_alias_returns_none_for_unknown_model():
router = _router_with_two_deployments([False, False])
assert router.get_deployment_model_for_alias(model_id="does-not-exist") is None
def test_get_deployment_model_for_alias_returns_none_for_blocked_deployment():
router = _router_with_two_deployments([True, False])
assert router.get_deployment_model_for_alias(model_id="dep-0") is None
assert router.get_deployment_model_for_alias(model_id="dep-1") == "openai/gpt-4o-1"
def test_resolve_unblocked_deployment_resolves_alias_id_and_wildcard():
"""
_resolve_unblocked_deployment underpins both the credential resolver and the