""" Tests for litellm/proxy/management_endpoints/common_utils.py Covers the fix for GitHub issue #20304: Empty guardrails/policies arrays sent by the UI should NOT trigger the enterprise (premium) license check, but should still be applied so that users can intentionally clear previously-set fields. """ from datetime import datetime, timezone from types import SimpleNamespace from typing import Final from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi import HTTPException from pydantic import BaseModel from litellm import Router from litellm.proxy._types import ( LiteLLM_BudgetTable, LiteLLM_OrganizationMembershipTable, LiteLLM_TeamTable, LiteLLM_UserTable, LitellmUserRoles, Member, UserAPIKeyAuth, ) from litellm.proxy.management_endpoints.common_utils import ( _has_non_empty_value, _org_admin_can_invite_user, _team_admin_can_invite_user, admin_can_invite_user, check_require_trace_id_caller_permission, set_object_metadata_field, update_metadata_fields, user_api_key_has_admin_view, user_has_admin_privileges, upsert_budget_and_membership, ) from litellm.types.utils import BudgetConfig @pytest.mark.asyncio @pytest.mark.parametrize( "budget_patch", [ {"tpm_limit": 200}, {"model_max_budget": {}, "tpm_limit": 5}, ], ) async def test_shared_default_clone_keeps_inheriting_updated_team_model_caps( budget_patch: dict[str, object], ) -> None: from litellm.proxy.auth.user_api_key_auth import _load_team_member_default_model_budget from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache, team_membership_reservation_cache_key from litellm.proxy._types import LiteLLM_TeamMembership source_caps = {"claude-sonnet-4-6": {"max_budget": 3.0, "budget_duration": "1d"}} default_caps = {"claude-sonnet-4-6": {"max_budget": 1.0, "budget_duration": "1d"}} budget_table = SimpleNamespace( find_unique=AsyncMock( return_value=SimpleNamespace( model_dump=MagicMock( return_value={ "budget_id": "default-budget", "model_max_budget": source_caps, "tpm_limit": 100, } ) ) ), create=AsyncMock(return_value=SimpleNamespace(budget_id="private-budget")), ) membership_table = SimpleNamespace(upsert=AsyncMock()) tx = SimpleNamespace(litellm_budgettable=budget_table, litellm_teammembership=membership_table) token = UserAPIKeyAuth(user_id="member-1", team_id="team-1") await upsert_budget_and_membership( tx, team_id="team-1", user_id="member-1", existing_budget_id="default-budget", user_api_key_dict=token, budget_patch=budget_patch, team_default_budget_id="default-budget", ) budget_table.create.assert_awaited_once() membership_table.upsert.assert_awaited_once() assert budget_table.create.await_args.kwargs["data"]["tpm_limit"] == budget_patch["tpm_limit"] model_max_budget: Final = budget_table.create.await_args.kwargs["data"]["model_max_budget"] assert isinstance(model_max_budget, str) assert model_max_budget == "{}" membership = LiteLLM_TeamMembership( user_id="member-1", team_id="team-1", budget_id="private-budget", litellm_budget_table=LiteLLM_BudgetTable(budget_id="private-budget", model_max_budget={}), ) user_api_key_cache: Final = UserApiKeyCache() await user_api_key_cache.async_set_cache( key=team_membership_reservation_cache_key(user_id="member-1", team_id="team-1"), value=membership, model_type=LiteLLM_TeamMembership, ) await user_api_key_cache.async_set_cache( key="team_member_default_budget:default-budget", value=LiteLLM_BudgetTable(budget_id="default-budget", model_max_budget=default_caps), model_type=LiteLLM_BudgetTable, ) team_object = SimpleNamespace( team_id="team-1", metadata={"team_member_budget_id": "default-budget"}, ) await _load_team_member_default_model_budget( user_api_key_auth_obj=token, team_object=team_object, prisma_client=MagicMock(), user_api_key_cache=user_api_key_cache, skip_budget_checks=False, proxy_logging_obj=None, ) assert token.team_member_model_max_budget == default_caps @pytest.mark.asyncio async def test_empty_model_budget_patch_on_team_default_keeps_membership_attached() -> None: tx: Final = MagicMock() tx.litellm_budgettable.find_unique = AsyncMock( return_value=SimpleNamespace( model_dump=MagicMock( return_value={ "budget_id": "default-budget", "max_budget": 10.0, "model_max_budget": {"claude-sonnet-4-6": {"max_budget": 1.0}}, } ) ) ) tx.litellm_budgettable.create = AsyncMock() tx.litellm_budgettable.update = AsyncMock() tx.litellm_teammembership.upsert = AsyncMock() tx.litellm_teammembership.update = AsyncMock() await upsert_budget_and_membership( tx, team_id="team-1", user_id="member-1", existing_budget_id="default-budget", user_api_key_dict=UserAPIKeyAuth(user_id="admin"), budget_patch={"model_max_budget": {}}, team_default_budget_id="default-budget", ) tx.litellm_budgettable.create.assert_not_awaited() tx.litellm_budgettable.update.assert_not_awaited() tx.litellm_teammembership.upsert.assert_not_awaited() tx.litellm_teammembership.update.assert_not_awaited() class TestUpdateMetadataFieldsEmptyCollections: @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_empty_list_does_not_trigger_premium_check(self, mock_premium_check): """Empty lists for premium fields must not trigger the premium check.""" updated_kv = { "team_id": "test-team", "guardrails": [], "policies": [], "logging": [], } update_metadata_fields(updated_kv=updated_kv) mock_premium_check.assert_not_called() @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_empty_list_still_updates_metadata(self, mock_premium_check): """ Empty lists must still be moved into metadata so users can clear previously-set fields (e.g. remove all guardrails). """ updated_kv = { "team_id": "test-team", "guardrails": [], "policies": [], } update_metadata_fields(updated_kv=updated_kv) # The fields should have been moved into metadata assert ( "guardrails" not in updated_kv ), "guardrails should be popped from top-level" assert "policies" not in updated_kv, "policies should be popped from top-level" assert updated_kv["metadata"]["guardrails"] == [] assert updated_kv["metadata"]["policies"] == [] @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_empty_dict_does_not_trigger_premium_check(self, mock_premium_check): """Empty dicts for premium fields must not trigger the premium check.""" updated_kv = { "team_id": "test-team", "secret_manager_settings": {}, } update_metadata_fields(updated_kv=updated_kv) mock_premium_check.assert_not_called() @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_empty_dict_still_updates_metadata(self, mock_premium_check): """ Empty dicts must still be moved into metadata so users can clear previously-set fields. """ updated_kv = { "team_id": "test-team", "secret_manager_settings": {}, } update_metadata_fields(updated_kv=updated_kv) assert ( "secret_manager_settings" not in updated_kv ), "secret_manager_settings should be popped from top-level" assert updated_kv["metadata"]["secret_manager_settings"] == {} @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_none_value_does_not_trigger_premium_check(self, mock_premium_check): """None values for premium fields should be silently ignored.""" updated_kv = { "team_id": "test-team", "guardrails": None, "policies": None, } update_metadata_fields(updated_kv=updated_kv) mock_premium_check.assert_not_called() @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_absent_fields_do_not_trigger_premium_check(self, mock_premium_check): """Fields not present in the dict should not trigger premium check.""" updated_kv = { "team_id": "test-team", "team_alias": "example-team", } update_metadata_fields(updated_kv=updated_kv) mock_premium_check.assert_not_called() @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_non_empty_list_triggers_premium_check(self, mock_premium_check): """Non-empty lists for premium fields should trigger the premium check.""" updated_kv = { "team_id": "test-team", "guardrails": ["my-guardrail"], } update_metadata_fields(updated_kv=updated_kv) mock_premium_check.assert_called() @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_non_empty_value_triggers_premium_check(self, mock_premium_check): """Non-empty string values for premium fields should trigger the premium check.""" updated_kv = { "team_id": "test-team", "tags": ["production"], } update_metadata_fields(updated_kv=updated_kv) mock_premium_check.assert_called() @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_non_empty_list_updates_metadata(self, mock_premium_check): """Non-empty lists should be moved into metadata.""" updated_kv = { "team_id": "test-team", "guardrails": ["my-guardrail"], } update_metadata_fields(updated_kv=updated_kv) assert "guardrails" not in updated_kv assert updated_kv["metadata"]["guardrails"] == ["my-guardrail"] @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_false_boolean_does_not_trigger_premium_check(self, mock_premium_check): """ Regression #30285: /team/update sends disable_global_guardrails=False (the UI's unchanged default). A falsy boolean must not trigger the premium check, so non-premium users are not wrongly 403'd. """ updated_kv = {"team_id": "test-team", "disable_global_guardrails": False} update_metadata_fields(updated_kv=updated_kv) mock_premium_check.assert_not_called() @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_false_boolean_still_updates_metadata(self, mock_premium_check): """A falsy boolean must still be moved into metadata so it persists.""" updated_kv = {"team_id": "test-team", "disable_global_guardrails": False} update_metadata_fields(updated_kv=updated_kv) assert "disable_global_guardrails" not in updated_kv assert updated_kv["metadata"]["disable_global_guardrails"] is False @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_true_boolean_triggers_premium_check(self, mock_premium_check): """Control: enabling the premium feature (True) still requires a license.""" updated_kv = {"team_id": "test-team", "disable_global_guardrails": True} update_metadata_fields(updated_kv=updated_kv) mock_premium_check.assert_called() @patch("litellm.proxy.management_endpoints.common_utils.premium_user_check") def test_ui_typical_payload_does_not_trigger_premium_check( self, mock_premium_check ): """ Simulate the exact payload the UI sends when no enterprise features are configured. This must NOT trigger the premium check. """ # This is the payload structure the UI sends (from issue #20304) updated_kv = { "team_id": "67848772-1a8b-4343-938c-17e60f1db860", "team_alias": "example-team", "models": ["gpt-4"], "metadata": { "guardrails": [], "logging": [], }, "policies": [], } update_metadata_fields(updated_kv=updated_kv) mock_premium_check.assert_not_called() class TestUserHasAdminView: """Tests for _user_has_admin_view function.""" @pytest.mark.parametrize( "user_role,expected", [ (LitellmUserRoles.PROXY_ADMIN, True), (LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, True), (LitellmUserRoles.INTERNAL_USER, False), (LitellmUserRoles.INTERNAL_USER_VIEW_ONLY, False), ], ) def test_user_has_admin_view_by_role(self, user_role, expected): """Parametrized test: admin roles return True, non-admin return False.""" mock_auth = MagicMock() mock_auth.user_role = user_role assert user_api_key_has_admin_view(mock_auth) == expected def test_user_has_admin_view_with_user_api_key_auth(self): """Test with actual UserAPIKeyAuth object.""" auth_admin = UserAPIKeyAuth( user_id="u1", api_key="sk-xxx", user_role=LitellmUserRoles.PROXY_ADMIN, ) auth_user = UserAPIKeyAuth( user_id="u2", api_key="sk-yyy", user_role=LitellmUserRoles.INTERNAL_USER, ) assert user_api_key_has_admin_view(auth_admin) is True assert user_api_key_has_admin_view(auth_user) is False def test_published_enterprise_import_of_team_admin_check_still_answers(): from litellm.proxy.management_endpoints.common_utils import _is_user_team_admin team = LiteLLM_TeamTable(team_id="t1", members_with_roles=[Member(user_id="admin", role="admin")]) assert _is_user_team_admin(UserAPIKeyAuth(user_id="admin"), team) is True assert _is_user_team_admin(UserAPIKeyAuth(user_id="outsider"), team) is False class TestOrgAdminCanInviteUser: """Tests for _org_admin_can_invite_user function.""" def _make_membership(self, org_id: str, user_role: str): now = datetime.now(timezone.utc) return LiteLLM_OrganizationMembershipTable( user_id="u", organization_id=org_id, user_role=user_role, created_at=now, updated_at=now, ) @pytest.mark.parametrize( "admin_orgs,target_orgs,expected", [ (["org1"], ["org1"], True), (["org1", "org2"], ["org2"], True), (["org1"], ["org2"], False), ([], ["org1"], False), (["org1"], [], False), ], ) def test_org_admin_can_invite_user_parametrized( self, admin_orgs, target_orgs, expected ): """Parametrized test: can invite when target is in org where admin has ORG_ADMIN role.""" admin_user = LiteLLM_UserTable( user_id="admin", organization_memberships=[ self._make_membership(oid, LitellmUserRoles.ORG_ADMIN.value) for oid in admin_orgs ], ) target_user = LiteLLM_UserTable( user_id="target", organization_memberships=[ self._make_membership(oid, LitellmUserRoles.INTERNAL_USER.value) for oid in target_orgs ], ) assert _org_admin_can_invite_user(admin_user, target_user) == expected def test_org_admin_can_invite_user_no_shared_org(self): """Test returns False when admin has no org admin role.""" admin_user = LiteLLM_UserTable( user_id="admin", organization_memberships=[ self._make_membership("org1", LitellmUserRoles.INTERNAL_USER.value), ], ) target_user = LiteLLM_UserTable( user_id="target", organization_memberships=[ self._make_membership("org1", LitellmUserRoles.INTERNAL_USER.value), ], ) assert _org_admin_can_invite_user(admin_user, target_user) is False class TestTeamAdminCanInviteUser: """Tests for _team_admin_can_invite_user async function.""" @pytest.mark.asyncio @pytest.mark.parametrize( "admin_teams,target_teams,user_is_admin_in,expected", [ (["t1"], ["t1"], ["t1"], True), (["t1", "t2"], ["t2"], ["t1", "t2"], True), (["t1"], ["t2"], ["t1"], False), ], ) async def test_team_admin_can_invite_user_parametrized( self, admin_teams, target_teams, user_is_admin_in, expected ): """Parametrized test: can invite when target shares a team where user is admin.""" mock_prisma = MagicMock() mock_auth = MagicMock() mock_auth.user_id = "admin" admin_user = LiteLLM_UserTable(user_id="admin", teams=admin_teams) target_user = LiteLLM_UserTable(user_id="target", teams=target_teams) def make_team(tid, is_admin): m = [{"user_id": "admin", "role": "admin"}] if is_admin else [] obj = MagicMock() obj.team_id = tid obj.model_dump = lambda: {"team_id": tid, "members_with_roles": m} return obj teams = [make_team(tid, tid in user_is_admin_in) for tid in admin_teams] mock_prisma.db.litellm_teamtable.find_many = AsyncMock(return_value=teams) result = await _team_admin_can_invite_user( user_api_key_dict=mock_auth, admin_user_obj=admin_user, target_user_obj=target_user, prisma_client=mock_prisma, ) assert result == expected @pytest.mark.asyncio async def test_team_admin_can_invite_user_no_shared_team(self): """Test returns False when admin and target share no team.""" mock_prisma = MagicMock() mock_auth = MagicMock() mock_auth.user_id = "admin" admin_user = LiteLLM_UserTable(user_id="admin", teams=[]) target_user = LiteLLM_UserTable(user_id="target", teams=["t1"]) result = await _team_admin_can_invite_user( user_api_key_dict=mock_auth, admin_user_obj=admin_user, target_user_obj=target_user, prisma_client=mock_prisma, ) assert result is False class TestUserHasAdminPrivileges: """Tests for _user_has_admin_privileges async function.""" @pytest.mark.asyncio async def test_proxy_admin_has_privileges(self): """Proxy admin always has admin privileges.""" auth = UserAPIKeyAuth( user_id="admin", api_key="sk-x", user_role=LitellmUserRoles.PROXY_ADMIN, ) result = await user_has_admin_privileges( user_api_key_dict=auth, prisma_client=None, ) assert result is True @pytest.mark.asyncio async def test_non_admin_no_prisma_returns_false(self): """Non-admin with no prisma connection has no privileges.""" auth = UserAPIKeyAuth( user_id="user1", api_key="sk-x", user_role=LitellmUserRoles.INTERNAL_USER, ) result = await user_has_admin_privileges( user_api_key_dict=auth, prisma_client=None, ) assert result is False class TestAdminCanInviteUser: """Tests for admin_can_invite_user async function.""" @pytest.mark.asyncio async def test_proxy_admin_can_invite_any_user(self): """Proxy admin can invite any user regardless of org/team.""" auth = UserAPIKeyAuth( user_id="admin", api_key="sk-x", user_role=LitellmUserRoles.PROXY_ADMIN, ) result = await admin_can_invite_user( target_user_id="any-user", user_api_key_dict=auth, prisma_client=None, ) assert result is True @pytest.mark.asyncio async def test_non_admin_cannot_invite_without_prisma(self): """Non-admin with no prisma cannot invite.""" auth = UserAPIKeyAuth( user_id="user1", api_key="sk-x", user_role=LitellmUserRoles.INTERNAL_USER, ) result = await admin_can_invite_user( target_user_id="other-user", user_api_key_dict=auth, prisma_client=None, ) assert result is False class TestSetObjectMetadataField: """Tests for _set_object_metadata_field function.""" @pytest.mark.parametrize( "field_name,value,should_call_premium", [ ("guardrails", ["g1"], True), ("model_rpm_limit", {"gpt-4": 10}, False), ], ) def test_set_object_metadata_field_parametrized( self, field_name, value, should_call_premium ): """Parametrized test: premium fields trigger _premium_user_check.""" team = LiteLLM_TeamTable(team_id="t1", metadata={}) with patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check" ) as mock_premium: set_object_metadata_field(team, field_name, value) if should_call_premium: mock_premium.assert_called_once() else: mock_premium.assert_not_called() assert team.metadata[field_name] == value def test_set_object_metadata_field_initializes_metadata_if_none(self): """Test initializes metadata dict when object has None.""" team = LiteLLM_TeamTable(team_id="t1", metadata=None) with patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check" ): set_object_metadata_field(team, "model_rpm_limit", {"x": 1}) assert team.metadata == {"model_rpm_limit": {"x": 1}} def test_mcp_rpm_limit_is_hoisted_into_metadata(self): """ Per-MCP-server rpm limits are stored in the metadata JSON column, not a dedicated DB column. The key/team management endpoints rely on LiteLLM_ManagementEndpoint_MetadataFields to move the request field into metadata; this regression guards that mcp_rpm_limit is in that list and round-trips through the same loop the endpoints use. """ from litellm.proxy._types import LiteLLM_ManagementEndpoint_MetadataFields assert "mcp_rpm_limit" in LiteLLM_ManagementEndpoint_MetadataFields from types import SimpleNamespace team = LiteLLM_TeamTable(team_id="t1", metadata={}) mcp_rpm_limit = {"github": 100} data = SimpleNamespace(mcp_rpm_limit=mcp_rpm_limit) with patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check" ): for field in LiteLLM_ManagementEndpoint_MetadataFields: if getattr(data, field, None) is not None: set_object_metadata_field(team, field, getattr(data, field)) assert team.metadata["mcp_rpm_limit"] == mcp_rpm_limit class TestRequireCallerUserIdForNonAdmin: """ Security regression: service-account keys (user_id=None) must not bypass the non-admin scoping branch on analytics endpoints. """ def test_returns_user_id_when_present(self): from litellm.proxy.management_endpoints.common_utils import ( require_caller_user_id_for_non_admin, ) key_dict = UserAPIKeyAuth( user_id="user-abc", user_role=LitellmUserRoles.INTERNAL_USER, ) assert require_caller_user_id_for_non_admin(key_dict) == "user-abc" def test_raises_403_when_user_id_is_none(self): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( require_caller_user_id_for_non_admin, ) # Simulates a service-account key (user_id forced to None at key creation) service_account_key = UserAPIKeyAuth( user_id=None, user_role=LitellmUserRoles.INTERNAL_USER, ) with pytest.raises(HTTPException) as exc_info: require_caller_user_id_for_non_admin(service_account_key) assert exc_info.value.status_code == 403 assert "Service-account keys" in str(exc_info.value.detail) class TestValidateFiniteSpend: """`validate_finite_spend` rejects NaN/±inf so a non-finite spend cannot bypass `spend >= max_budget` enforcement (NaN/-inf compare false).""" def test_none_is_allowed(self): from litellm.proxy.management_endpoints.common_utils import ( validate_finite_spend, ) assert validate_finite_spend(None) is None def test_finite_value_is_allowed(self): from litellm.proxy.management_endpoints.common_utils import ( validate_finite_spend, ) assert validate_finite_spend(0.0) is None assert validate_finite_spend(12.5) is None # Negative spend is intentionally allowed. Admins may set a negative # spend counter to grant an entity extra allowance for the current # budget period only (e.g. a large one-time spend grant), effectively # raising their headroom without raising the recurring budget ceiling. # Future changes should continue to allow negative spend counters. assert validate_finite_spend(-50.0) is None @pytest.mark.parametrize("bad", [float("nan"), float("inf"), float("-inf")]) def test_non_finite_is_rejected(self, bad): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( validate_finite_spend, ) with pytest.raises(HTTPException) as exc_info: validate_finite_spend(bad) assert exc_info.value.status_code == 400 class TestValidateFiniteSpendErrorDetail: """The 400 for non-finite spend must carry the exact {"error": } body.""" def test_rejection_detail_is_exact(self): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( validate_finite_spend, ) with pytest.raises(HTTPException) as exc_info: validate_finite_spend(float("nan")) assert exc_info.value.detail == { "error": "spend must be a finite number. Received: nan" } class TestValidateBudgetDuration: """`validate_budget_duration` keeps durations that never advance out of the database. A duration of "0s" resolves to a reset time of now, so the row is due again the instant it is written. The reset job re-reads such rows on every tick and, once one tenant owns enough of them, they fill each batch and starve every other tenant's reset. """ def test_none_is_allowed(self): from litellm.proxy.management_endpoints.common_utils import ( validate_budget_duration, ) assert validate_budget_duration(None) is None @pytest.mark.parametrize("duration", ["30s", "5m", "1h", "1d", "7d", "30d", "1mo"]) def test_positive_durations_are_allowed(self, duration): from litellm.proxy.management_endpoints.common_utils import ( validate_budget_duration, ) assert validate_budget_duration(duration) is None @pytest.mark.parametrize("duration", ["0s", "0m", "0h", "0d", "-5m", "abc", ""]) def test_non_advancing_durations_are_rejected(self, duration): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( validate_budget_duration, ) with pytest.raises(HTTPException) as exc_info: validate_budget_duration(duration) assert exc_info.value.status_code == 400 def test_rejection_detail_is_exact(self): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( validate_budget_duration, ) with pytest.raises(HTTPException) as exc_info: validate_budget_duration("0s") assert exc_info.value.detail == { "error": "Invalid budget_duration '0s'. Use a format like '1h', '24h', '7d', or '30d'." } class TestRequireCallerUserIdErrorDetail: """The 403 for a service-account key must carry the exact error body.""" def test_rejection_detail_is_exact(self): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( require_caller_user_id_for_non_admin, ) service_account_key = UserAPIKeyAuth( user_id=None, user_role=LitellmUserRoles.INTERNAL_USER, ) with pytest.raises(HTTPException) as exc_info: require_caller_user_id_for_non_admin(service_account_key) assert exc_info.value.detail == { "error": "Service-account keys cannot query user analytics. Use a user-bound key, or call as a proxy admin." } class TestCheckPassthroughRoutesCallerPermission: """Only proxy admins may set allowed_passthrough_routes (top-level or under metadata); non-admins get a 403 naming the entity.""" def _non_admin(self): return UserAPIKeyAuth( user_id="u1", api_key="sk-x", user_role=LitellmUserRoles.INTERNAL_USER ) def test_top_level_routes_rejected_with_default_entity(self): from fastapi import HTTPException from pydantic import BaseModel from litellm.proxy.management_endpoints.common_utils import ( check_passthrough_routes_caller_permission, ) class _RouteData(BaseModel): allowed_passthrough_routes: list | None = None metadata: dict | None = None data = _RouteData(allowed_passthrough_routes=["/v1/foo"]) with pytest.raises(HTTPException) as exc_info: check_passthrough_routes_caller_permission(data, self._non_admin()) assert exc_info.value.status_code == 403 assert exc_info.value.detail == { "error": "Only proxy admins can set `allowed_passthrough_routes` on a key." } def test_metadata_routes_rejected_with_default_entity(self): from fastapi import HTTPException from pydantic import BaseModel from litellm.proxy.management_endpoints.common_utils import ( check_passthrough_routes_caller_permission, ) class _RouteData(BaseModel): allowed_passthrough_routes: list | None = None metadata: dict | None = None data = _RouteData(metadata={"allowed_passthrough_routes": ["/v1/foo"]}) with pytest.raises(HTTPException) as exc_info: check_passthrough_routes_caller_permission(data, self._non_admin()) assert exc_info.value.detail == { "error": "Only proxy admins can set `metadata.allowed_passthrough_routes` on a key." } def test_tolerates_data_missing_passthrough_and_metadata_fields(self): from pydantic import BaseModel from litellm.proxy.management_endpoints.common_utils import ( check_passthrough_routes_caller_permission, ) class _Bare(BaseModel): unrelated: str = "x" assert check_passthrough_routes_caller_permission(_Bare(), self._non_admin()) is None @pytest.mark.parametrize( "kwargs, field", [ ({"denied_passthrough_routes": ["/v1/foo"]}, "denied_passthrough_routes"), ({"metadata": {"denied_passthrough_routes": ["/v1/foo"]}}, "metadata.denied_passthrough_routes"), ], ) def test_denied_routes_rejected_for_non_admin(self, kwargs: dict[str, object], field: str) -> None: from fastapi import HTTPException from pydantic import BaseModel from litellm.proxy.management_endpoints.common_utils import ( check_passthrough_routes_caller_permission, ) class _RouteData(BaseModel): denied_passthrough_routes: list[str] | None = None metadata: dict[str, object] | None = None with pytest.raises(HTTPException) as exc_info: check_passthrough_routes_caller_permission( _RouteData.model_validate(kwargs), self._non_admin(), entity="team" ) assert exc_info.value.detail == {"error": f"Only proxy admins can set `{field}` on a team."} class _DenyRouteData(BaseModel): denied_passthrough_routes: list[str] | None = None metadata: dict[str, object] | None = None max_budget: float | None = None _EXISTING_DENY: Final = {"denied_passthrough_routes": ["/v1/foo"]} class TestDeniedPassthroughRoutesCallerPermission: @pytest.mark.parametrize( "kwargs, field", [ ({"denied_passthrough_routes": []}, "denied_passthrough_routes"), ({"denied_passthrough_routes": ["/v1/other"]}, "denied_passthrough_routes"), ({"metadata": {"team": "core"}}, "metadata.denied_passthrough_routes"), ({"metadata": None}, "metadata.denied_passthrough_routes"), ], ids=["cleared", "replaced", "dropped-by-metadata-replace", "dropped-by-null-metadata"], ) def test_non_admin_cannot_change_an_existing_deny_list(self, kwargs: dict[str, object], field: str) -> None: from litellm.proxy.management_endpoints.common_utils import check_passthrough_routes_caller_permission with pytest.raises(HTTPException) as exc_info: check_passthrough_routes_caller_permission( _DenyRouteData.model_validate(kwargs), UserAPIKeyAuth(user_id="u1", user_role=LitellmUserRoles.INTERNAL_USER), existing_metadata=_EXISTING_DENY, ) assert exc_info.value.status_code == 403 assert exc_info.value.detail == {"error": f"Only proxy admins can set `{field}` on a key."} @pytest.mark.parametrize( "kwargs", [ {"denied_passthrough_routes": ["/v1/foo"]}, {"metadata": {"team": "core", "denied_passthrough_routes": ["/v1/foo"]}}, {"max_budget": 10.0}, ], ids=["resent-top-level", "resent-in-metadata", "unrelated-field"], ) def test_non_admin_may_leave_an_existing_deny_list_unchanged(self, kwargs: dict[str, object]) -> None: from litellm.proxy.management_endpoints.common_utils import check_passthrough_routes_caller_permission check_passthrough_routes_caller_permission( _DenyRouteData.model_validate(kwargs), UserAPIKeyAuth(user_id="u1", user_role=LitellmUserRoles.INTERNAL_USER), existing_metadata=_EXISTING_DENY, ) def test_non_admin_may_send_null_metadata_when_no_deny_list_exists(self) -> None: from litellm.proxy.management_endpoints.common_utils import check_passthrough_routes_caller_permission check_passthrough_routes_caller_permission( _DenyRouteData(metadata=None), UserAPIKeyAuth(user_id="u1", user_role=LitellmUserRoles.INTERNAL_USER), existing_metadata={"team": "core"}, ) def test_malformed_metadata_deny_entries_are_rejected_even_for_proxy_admins(self) -> None: from litellm.proxy.management_endpoints.common_utils import check_passthrough_routes_caller_permission with pytest.raises(HTTPException) as exc_info: check_passthrough_routes_caller_permission( _DenyRouteData(metadata={"denied_passthrough_routes": [123, None]}), UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN), ) assert exc_info.value.status_code == 400 assert exc_info.value.detail == { "error": "`metadata.denied_passthrough_routes` must be a list of route strings." } class TestCheckDisableGlobalGuardrailsCallerPermission: """Only proxy admins may set disable_global_guardrails (top-level or under metadata); non-admins get a 403 naming the entity.""" def _non_admin(self): return UserAPIKeyAuth( user_id="u1", api_key="sk-x", user_role=LitellmUserRoles.INTERNAL_USER ) def _admin(self): return UserAPIKeyAuth( user_id="u2", api_key="sk-y", user_role=LitellmUserRoles.PROXY_ADMIN ) def test_top_level_flag_rejected_with_default_entity(self): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( check_disable_global_guardrails_caller_permission, ) with pytest.raises(HTTPException) as exc_info: check_disable_global_guardrails_caller_permission(True, None, self._non_admin()) assert exc_info.value.status_code == 403 assert exc_info.value.detail == {"error": "Only proxy admins can set `disable_global_guardrails` on a key."} def test_metadata_flag_rejected_with_default_entity(self): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( check_disable_global_guardrails_caller_permission, ) with pytest.raises(HTTPException) as exc_info: check_disable_global_guardrails_caller_permission( None, {"disable_global_guardrails": True}, self._non_admin() ) assert exc_info.value.status_code == 403 assert exc_info.value.detail == {"error": "Only proxy admins can set `disable_global_guardrails` on a key."} def test_explicit_false_with_metadata_true_is_rejected(self): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( check_disable_global_guardrails_caller_permission, ) with pytest.raises(HTTPException) as exc_info: check_disable_global_guardrails_caller_permission( False, {"disable_global_guardrails": True}, self._non_admin() ) assert exc_info.value.status_code == 403 assert exc_info.value.detail == {"error": "Only proxy admins can set `disable_global_guardrails` on a key."} def test_rejection_names_the_team_entity(self): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( check_disable_global_guardrails_caller_permission, ) with pytest.raises(HTTPException) as exc_info: check_disable_global_guardrails_caller_permission(True, None, self._non_admin(), entity="team") assert exc_info.value.detail == {"error": "Only proxy admins can set `disable_global_guardrails` on a team."} def test_false_and_absent_flag_do_not_raise(self): from litellm.proxy.management_endpoints.common_utils import ( check_disable_global_guardrails_caller_permission, ) non_admin = self._non_admin() assert check_disable_global_guardrails_caller_permission(False, None, non_admin) is None assert check_disable_global_guardrails_caller_permission(None, None, non_admin) is None assert check_disable_global_guardrails_caller_permission(None, {}, non_admin) is None assert ( check_disable_global_guardrails_caller_permission(None, {"disable_global_guardrails": False}, non_admin) is None ) def test_unchanged_stored_flag_does_not_raise(self): """Re-sending a flag that is already stored is not an opt-out.""" from litellm.proxy.management_endpoints.common_utils import ( check_disable_global_guardrails_caller_permission, ) non_admin = self._non_admin() assert ( check_disable_global_guardrails_caller_permission( True, {"disable_global_guardrails": True}, non_admin, existing_metadata={"disable_global_guardrails": True}, ) is None ) def test_stored_false_does_not_exempt(self): from fastapi import HTTPException from litellm.proxy.management_endpoints.common_utils import ( check_disable_global_guardrails_caller_permission, ) with pytest.raises(HTTPException) as exc_info: check_disable_global_guardrails_caller_permission( True, None, self._non_admin(), existing_metadata={"disable_global_guardrails": False}, ) assert exc_info.value.status_code == 403 assert exc_info.value.detail == {"error": "Only proxy admins can set `disable_global_guardrails` on a key."} def test_proxy_admin_may_set_the_flag(self): from litellm.proxy.management_endpoints.common_utils import ( check_disable_global_guardrails_caller_permission, ) assert ( check_disable_global_guardrails_caller_permission(True, {"disable_global_guardrails": True}, self._admin()) is None ) @pytest.mark.parametrize( ("require_trace_id", "metadata", "metadata_sent", "existing_metadata"), [ pytest.param(True, None, False, None, id="create-with-top-level-true"), pytest.param(False, None, False, {"require_trace_id": True}, id="clear-stored-value-top-level"), pytest.param(None, {}, True, {"require_trace_id": True}, id="replace-metadata-omitting-stored-value"), pytest.param( None, None, True, {"require_trace_id": True}, id="explicit-null-metadata-clears-stored-value", ), pytest.param( False, {"require_trace_id": True}, True, {"require_trace_id": True}, id="top-level-false-wins-over-metadata", ), ], ) def test_non_proxy_admin_cannot_change_require_trace_id( require_trace_id: bool | None, metadata: dict[str, object] | None, metadata_sent: bool, existing_metadata: dict[str, object] | None, ) -> None: exc_info: Final[pytest.ExceptionInfo[HTTPException]] with pytest.raises(HTTPException) as exc_info: check_require_trace_id_caller_permission( require_trace_id, metadata, UserAPIKeyAuth(user_id="u1", user_role=LitellmUserRoles.INTERNAL_USER), metadata_sent=metadata_sent, existing_metadata=existing_metadata, ) assert exc_info.value.status_code == 403 assert exc_info.value.detail == {"error": "Only proxy admins can set `require_trace_id` on a team."} @pytest.mark.parametrize( ("require_trace_id", "metadata", "metadata_sent", "existing_metadata"), [ pytest.param(True, {"require_trace_id": True}, True, {"require_trace_id": True}, id="resend-stored-value"), pytest.param(None, {}, True, None, id="metadata-without-stored-value"), pytest.param( True, {}, True, {"require_trace_id": True}, id="top-level-true-with-empty-metadata-keeps-stored-value", ), pytest.param(None, None, False, {"require_trace_id": True}, id="metadata-not-sent"), ], ) def test_non_proxy_admin_can_preserve_or_omit_require_trace_id( require_trace_id: bool | None, metadata: dict[str, object] | None, metadata_sent: bool, existing_metadata: dict[str, object] | None, ) -> None: assert ( check_require_trace_id_caller_permission( require_trace_id, metadata, UserAPIKeyAuth(user_id="u1", user_role=LitellmUserRoles.INTERNAL_USER), metadata_sent=metadata_sent, existing_metadata=existing_metadata, ) is None ) def test_proxy_admin_can_flip_stored_require_trace_id() -> None: assert ( check_require_trace_id_caller_permission( False, None, UserAPIKeyAuth(user_id="u2", user_role=LitellmUserRoles.PROXY_ADMIN), metadata_sent=False, existing_metadata={"require_trace_id": True}, ) is None ) class TestTeamMemberHasPermission: def test_requires_caller_to_be_a_team_member(self): from litellm.proxy.management_endpoints.common_utils import ( team_member_has_permission, ) team = LiteLLM_TeamTable( team_id="t1", team_member_permissions=["/key/generate"], members_with_roles=[Member(user_id="someone-else", role="user")], ) key = UserAPIKeyAuth( user_id="u1", api_key="sk-x", user_role=LitellmUserRoles.INTERNAL_USER ) assert team_member_has_permission(key, team, "/key/generate") is False class TestUserHasAdminPrivilegesGuard: @pytest.mark.asyncio async def test_no_user_lookup_when_prisma_is_none(self): """With no DB the guard short-circuits before any user lookup.""" auth = UserAPIKeyAuth( user_id="user1", api_key="sk-x", user_role=LitellmUserRoles.INTERNAL_USER ) mock_get_user = AsyncMock(return_value=None) with patch("litellm.proxy.auth.auth_checks.get_user_object", mock_get_user): result = await user_has_admin_privileges( user_api_key_dict=auth, prisma_client=None ) assert result is False mock_get_user.assert_not_called() @pytest.mark.asyncio async def test_org_admin_membership_grants_privileges(self): """With DB + user_id present, an ORG_ADMIN membership yields True.""" auth = UserAPIKeyAuth( user_id="user1", api_key="sk-x", user_role=LitellmUserRoles.INTERNAL_USER ) now = datetime.now(timezone.utc) user_obj = LiteLLM_UserTable( user_id="user1", organization_memberships=[ LiteLLM_OrganizationMembershipTable( user_id="user1", organization_id="org1", user_role=LitellmUserRoles.ORG_ADMIN.value, created_at=now, updated_at=now, ) ], ) mock_get_user = AsyncMock(return_value=user_obj) with patch("litellm.proxy.auth.auth_checks.get_user_object", mock_get_user): result = await user_has_admin_privileges( user_api_key_dict=auth, prisma_client=MagicMock() ) assert result is True class TestAdminCanInviteUserGuard: @pytest.mark.asyncio async def test_no_user_lookup_when_prisma_is_none(self): auth = UserAPIKeyAuth( user_id="admin1", api_key="sk-x", user_role=LitellmUserRoles.INTERNAL_USER ) mock_get_user = AsyncMock(return_value=None) with patch("litellm.proxy.auth.auth_checks.get_user_object", mock_get_user): result = await admin_can_invite_user( target_user_id="target1", user_api_key_dict=auth, prisma_client=None, ) assert result is False mock_get_user.assert_not_called() @pytest.mark.asyncio async def test_org_admin_can_invite_user_in_shared_org(self): now = datetime.now(timezone.utc) auth = UserAPIKeyAuth( user_id="admin1", api_key="sk-x", user_role=LitellmUserRoles.INTERNAL_USER ) def membership(role): return LiteLLM_OrganizationMembershipTable( user_id="x", organization_id="org1", user_role=role, created_at=now, updated_at=now, ) admin_obj = LiteLLM_UserTable( user_id="admin1", organization_memberships=[membership(LitellmUserRoles.ORG_ADMIN.value)], ) target_obj = LiteLLM_UserTable( user_id="target1", organization_memberships=[membership(LitellmUserRoles.INTERNAL_USER.value)], ) mock_get_user = AsyncMock(side_effect=[admin_obj, target_obj]) with patch("litellm.proxy.auth.auth_checks.get_user_object", mock_get_user): result = await admin_can_invite_user( target_user_id="target1", user_api_key_dict=auth, prisma_client=MagicMock(), ) assert result is True class TestTeamAdminCanInviteUserQuery: @pytest.mark.asyncio async def test_find_many_queries_admin_teams_with_exact_where(self): mock_prisma = MagicMock() mock_auth = MagicMock() mock_auth.user_id = "admin" admin_user = LiteLLM_UserTable(user_id="admin", teams=["t1", "t2"]) target_user = LiteLLM_UserTable(user_id="target", teams=["t2"]) def make_team(tid): obj = MagicMock() obj.team_id = tid obj.model_dump = lambda: { "team_id": tid, "members_with_roles": [{"user_id": "admin", "role": "admin"}], } return obj find_many = AsyncMock(return_value=[make_team("t1"), make_team("t2")]) mock_prisma.db.litellm_teamtable.find_many = find_many await _team_admin_can_invite_user( user_api_key_dict=mock_auth, admin_user_obj=admin_user, target_user_obj=target_user, prisma_client=mock_prisma, ) find_many.assert_awaited_once_with(where={"team_id": {"in": ["t1", "t2"]}}) class TestSetObjectMetadataFieldPremiumArg: def test_premium_check_receives_the_field_name(self): team = LiteLLM_TeamTable(team_id="t1", metadata={}) with patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check" ) as mock_premium: set_object_metadata_field(team, "guardrails", ["g1"]) mock_premium.assert_called_once_with("guardrails") class TestUpdateMetadataFieldMove: def test_none_valued_field_is_not_moved_into_metadata(self): """A None value must leave the field untouched (guard requires non-None).""" from litellm.proxy.management_endpoints.common_utils import ( _update_metadata_field, ) updated_kv = {"guardrails": None} _update_metadata_field(updated_kv=updated_kv, field_name="guardrails") assert updated_kv == {"guardrails": None} def test_set_premium_field_is_moved_into_metadata(self): updated_kv = {"guardrails": ["g1"]} with patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check" ): update_metadata_fields(updated_kv) assert "guardrails" not in updated_kv assert updated_kv["metadata"]["guardrails"] == ["g1"] class TestHasNonEmptyValue: """Tests for the _has_non_empty_value helper.""" def test_none_is_empty(self): assert _has_non_empty_value(None) is False def test_empty_list_is_empty(self): assert _has_non_empty_value([]) is False def test_empty_string_is_empty(self): assert _has_non_empty_value("") is False def test_blank_string_is_empty(self): assert _has_non_empty_value(" ") is False def test_non_empty_list_has_value(self): assert _has_non_empty_value(["policy-a"]) is True def test_non_empty_string_has_value(self): assert _has_non_empty_value("30d") is True def test_dict_has_value(self): assert _has_non_empty_value({"key": "val"}) is True def test_empty_dict_has_value(self): # empty dict is not None/list/str, so it counts as non-empty assert _has_non_empty_value({}) is True class TestUpdateMetadataFieldsPremiumCheck: """ Tests that _update_metadata_fields skips premium user checks for empty values but still enforces them for real values. Issue: The UI sends the full form on every team update, including premium fields like `policies: []`. The backend was treating these empty values as premium feature usage and returning 403. """ @patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check", side_effect=Exception("Should not be called"), ) def test_empty_policies_skips_premium_check(self, mock_check): """policies: [] should NOT trigger premium user check.""" updated_kv = { "team_id": "team-123", "team_alias": "my-team", "policies": [], } update_metadata_fields(updated_kv) mock_check.assert_not_called() @patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check", side_effect=Exception("Should not be called"), ) def test_empty_guardrails_skips_premium_check(self, mock_check): """guardrails: [] should NOT trigger premium user check.""" updated_kv = { "team_id": "team-123", "guardrails": [], } update_metadata_fields(updated_kv) mock_check.assert_not_called() @patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check", side_effect=Exception("Should not be called"), ) def test_empty_string_team_member_key_duration_skips_premium_check( self, mock_check ): """team_member_key_duration: '' should NOT trigger premium user check.""" updated_kv = { "team_id": "team-123", "team_member_key_duration": "", } update_metadata_fields(updated_kv) mock_check.assert_not_called() @patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check", side_effect=Exception("Should not be called"), ) def test_full_ui_payload_with_empty_premium_fields_skips_premium_check( self, mock_check ): """A realistic UI payload with all empty premium fields should not 403.""" updated_kv = { "team_id": "team-123", "team_alias": "renamed-team", "models": ["gpt-4o"], "max_budget": 200, "policies": [], "guardrails": [], "logging": [], "team_member_key_duration": "", "prompts": [], } update_metadata_fields(updated_kv) mock_check.assert_not_called() @patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check", ) def test_non_empty_policies_triggers_premium_check(self, mock_check): """policies: ['real-policy'] SHOULD trigger premium user check.""" updated_kv = { "team_id": "team-123", "policies": ["real-policy"], } update_metadata_fields(updated_kv) mock_check.assert_called() @patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check", ) def test_non_empty_guardrails_triggers_premium_check(self, mock_check): """guardrails: ['my-guardrail'] SHOULD trigger premium user check.""" updated_kv = { "team_id": "team-123", "guardrails": ["my-guardrail"], } update_metadata_fields(updated_kv) mock_check.assert_called() @patch( "litellm.proxy.management_endpoints.common_utils.premium_user_check", ) def test_non_empty_team_member_key_duration_triggers_premium_check( self, mock_check ): """team_member_key_duration: '30d' SHOULD trigger premium user check.""" updated_kv = { "team_id": "team-123", "team_member_key_duration": "30d", } update_metadata_fields(updated_kv) mock_check.assert_called() @pytest.mark.asyncio @pytest.mark.parametrize("db_model,stored_name,owner,public_name,error", [ (False, None, None, None, None), (True, "group", None, None, None), (True, None, None, None, "Unknown deployment ID in router weights: id"), (False, "renamed", None, None, "Deployment id does not belong to model group group"), (False, None, "other-team", None, "Unknown deployment ID in router weights: id"), (True, "internal", "team", "group", None), (True, "group", "team", "public", "Deployment id does not belong to model group group"), (True, "group", None, "unrelated-public-name", None), ]) async def test_router_weights_validate_current_deployment_scope( db_model: bool, stored_name: str | None, owner: str | None, public_name: str | None, error: str | None, ) -> None: from litellm.proxy.management_endpoints.router_weights import validate_router_settings_weights info = {"team_id": owner, "team_public_model_name": public_name} router = Router(model_list=[{ "model_name": "group", "litellm_params": {"model": "openai/gpt-5.4-mini", "api_key": "test"}, "model_info": {"id": "id", "db_model": db_model, **info}, }]) rows = [SimpleNamespace(model_id="id", model_name=stored_name, model_info=info)] if stored_name else [] table = SimpleNamespace(find_many=AsyncMock(return_value=rows)) db = SimpleNamespace(db=SimpleNamespace(litellm_proxymodeltable=table)) validation = validate_router_settings_weights( {"weights": {"group": {"id": 1}}}, team_id="team", prisma_client=db, llm_router=router, ) if error: with pytest.raises(HTTPException, match=error) as exc: await validation assert exc.value.status_code == 400 assert exc.value.detail == error else: await validation @pytest.mark.parametrize( "model_max_budget, error", [ ({"gpt-4o": BudgetConfig(max_budget=-1.0, budget_duration="1d")}, "non-negative finite"), ({"gpt-4o": BudgetConfig(max_budget=float("inf"), budget_duration="1d")}, "non-negative finite"), ({"gpt-4o": BudgetConfig(max_budget=float("nan"), budget_duration="1d")}, "non-negative finite"), ({"gpt-4o": BudgetConfig(budget_duration="1d")}, "non-negative finite"), ({"gpt-4o": BudgetConfig(max_budget=5.0)}, "requires a budget_duration"), ({"gpt-4o": BudgetConfig(max_budget=5.0, budget_duration="fortnight")}, "budget_duration"), ({" ": BudgetConfig(max_budget=5.0, budget_duration="1d")}, "non-empty model names"), ({"gpt-4o": BudgetConfig(max_budget=5.0, budget_duration="1d", tpm_limit=1000)}, "not enforced on a team"), ({"gpt-4o": BudgetConfig(max_budget=5.0, budget_duration="1d", rpm_limit=10)}, "not enforced on a team"), ], ids=["negative", "inf", "nan", "no_cap", "no_duration", "bad_duration", "blank_model", "tpm_limit", "rpm_limit"], ) def test_validate_team_model_max_budget_rejects_unenforceable_entries(model_max_budget, error) -> None: from litellm.proxy.management_endpoints.common_utils import validate_team_model_max_budget with pytest.raises(HTTPException) as exc: validate_team_model_max_budget(model_max_budget=model_max_budget, premium_user=True) assert exc.value.status_code == 400 assert error in exc.value.detail["error"] def test_validate_team_model_max_budget_accepts_a_zero_cap_and_prefixed_models() -> None: from litellm.proxy.management_endpoints.common_utils import validate_team_model_max_budget assert ( validate_team_model_max_budget( model_max_budget={ "gpt-4o": BudgetConfig(max_budget=0.0, budget_duration="1d"), "openai/gpt-4o-mini": BudgetConfig(max_budget=2.5, budget_duration="30d"), }, premium_user=True, ) is None ) def test_validate_team_model_max_budget_is_license_gated_only_when_set() -> None: from litellm.proxy.management_endpoints.common_utils import validate_team_model_max_budget validate_team_model_max_budget(model_max_budget=None, premium_user=False) validate_team_model_max_budget(model_max_budget={}, premium_user=False) with pytest.raises(HTTPException) as exc: validate_team_model_max_budget( model_max_budget={"gpt-4o": BudgetConfig(max_budget=1.0, budget_duration="1d")}, premium_user=False ) assert exc.value.status_code == 403