scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-10 03:28:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 84

test(router): cover fusion adapter entrypoints

Browse source
This commit is contained in:
moe-berri 2026-09-03 13:55:25 -07:00
parent 2852fb846a
commit 5746db7081
2 changed files with 6 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
litellm/proxy/hooks/proxy_track_cost_callback.py
View file

@ -367,8 +367,7 @@ class _ProxyDBLogger(CustomLogger):
kwargs=kwargs,
)
budget_counter_response_cost: Final = (
float(response_cost)
+ float(budget_reservation.get(FUSION_BUDGET_ACCUMULATED_COST_KEY) or 0.0)
float(response_cost) + float(budget_reservation.get(FUSION_BUDGET_ACCUMULATED_COST_KEY) or 0.0)
if budget_reservation is not None
and budget_reservation.get(FUSION_BUDGET_ACTIVE_KEY) is True
and metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY) == "fusion_continuation"
@ -727,9 +726,7 @@ async def _update_database_and_spend_counters(
token=user_api_key,
team_id=team_id,
user_id=user_id,
response_cost=(
budget_counter_response_cost if budget_counter_response_cost is not None else response_cost
),
response_cost=(budget_counter_response_cost if budget_counter_response_cost is not None else response_cost),
org_id=org_id,
budget_reservation=budget_reservation,
end_user_id=end_user_id,

7
tests/test_litellm/test_fusion_router.py
View file

@ -655,6 +655,7 @@ def _router_model_list() -> list[dict[str, object]]:
@pytest.mark.asyncio
async def test_router_registers_and_executes_fusion_deployment() -> None:
router = Router(model_list=_router_model_list())
assert router.get_configured_mode("fusion/test") is None
response = await router.acompletion(model="fusion/test", messages=[{"role": "user", "content": "Answer"}])
assert isinstance(response, ModelResponse)
assert response.choices[0].message.content == "Final"
@ -765,10 +766,10 @@ async def test_fusion_search_fails_closed_when_proxy_auth_context_is_missing(mon
@pytest.mark.asyncio
async def test_router_responses_and_anthropic_adapters_use_same_fusion_model() -> None:
router = Router(model_list=_router_model_list())
responses_result = await router.aresponses(model="fusion/test", input="Answer")
responses_result = await router._fusion_aware_aresponses(model="fusion/test", input="Answer")
assert responses_result.output[0].content[0].text == "Final"
assert inspect.iscoroutinefunction(router.aanthropic_messages)
anthropic_result = await router.aanthropic_messages(
anthropic_result = await router._fusion_aware_aanthropic_messages(
model="fusion/test", messages=[{"role": "user", "content": "Answer"}], max_tokens=256
)
assert anthropic_result["content"][0]["text"] == "Final"
@ -798,5 +799,5 @@ def test_sync_router_and_responses_support_nonstreaming_fusion() -> None:
response = router.completion(model="fusion/test", messages=[{"role": "user", "content": "Answer"}])
assert isinstance(response, ModelResponse)
assert response.choices[0].message.content == "Final"
responses_result = router.responses(model="fusion/test", input="Answer")
responses_result = router._fusion_aware_responses(model="fusion/test", input="Answer")
assert responses_result.output[0].content[0].text == "Final"