From b07db54b0a60c2caf7d14afcdf27419ed3e081b3 Mon Sep 17 00:00:00 2001 From: Fabrizio Siciliano Date: Tue, 7 Jul 2026 12:21:37 +0200 Subject: [PATCH 1/2] fix(proxy): naming convention fix for ui setting --- .../key_management_endpoints.py | 17 +- .../test_key_management_endpoints.py | 1418 +++++------------ 2 files changed, 413 insertions(+), 1022 deletions(-) diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index cef6ef54d81..daed7d83968 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -178,7 +178,13 @@ async def _enforce_email_prefix_on_key_alias( if key_alias is None: return None ui_settings = await get_ui_settings_cached() - if ui_settings.get("enforce_email_prefix_on_key_alias") is not True: + if "enforce_email_prefix_on_key_alias" in ui_settings: + is_enabled = ui_settings["enforce_email_prefix_on_key_alias"] is True + else: + from litellm.proxy.proxy_server import general_settings + + is_enabled = general_settings.get("enforce_email_prefix_on_key_alias") is True + if not is_enabled: return key_alias resolved_email = next( (email.strip() for email in (owner_email, fallback_email) if email is not None and email.strip()), @@ -1073,6 +1079,8 @@ async def _common_key_generation_helper( fallback_email=user_api_key_dict.user_email, ) + verbose_proxy_logger.warning(f'data_json["key_alias"]: {data_json["key_alias"]}') + _validate_key_alias_format(key_alias=data_json.get("key_alias", None)) await _enforce_unique_key_alias( @@ -2034,11 +2042,14 @@ async def prepare_key_update_data( owner_email=( await _get_user_email_by_id( prisma_client=prisma_client, - user_id=cast(Optional[str], non_default_values.get("user_id", existing_key_row.user_id)), + user_id=cast( + Optional[str], + non_default_values.get("user_id", existing_key_row.user_id), + ), ) ) or getattr(existing_key_row, "user_email", None), - fallback_email=user_api_key_dict.user_email if user_api_key_dict is not None else None, + fallback_email=(user_api_key_dict.user_email if user_api_key_dict is not None else None), ) return non_default_values diff --git a/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py index 3febaa38545..9bf51b6d83a 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py @@ -7,9 +7,7 @@ import yaml from fastapi.testclient import TestClient -sys.path.insert( - 0, os.path.abspath("../../../..") -) # Adds the parent directory to the system path +sys.path.insert(0, os.path.abspath("../../../..")) # Adds the parent directory to the system path from unittest.mock import AsyncMock, MagicMock, patch @@ -90,9 +88,7 @@ async def test_list_keys(): where_condition = mock_find_many.call_args.kwargs["where"] print(f"where_condition: {where_condition}") - assert json.dumps({"team_id": {"not": "litellm-dashboard"}}) in json.dumps( - where_condition - ) + assert json.dumps({"team_id": {"not": "litellm-dashboard"}}) in json.dumps(where_condition) @pytest.mark.asyncio @@ -170,9 +166,7 @@ async def test_list_keys_include_created_by_keys(): # Verify created_by condition only has the created_by filter (no other filters applied) # This is the current behavior - created_by keys don't inherit other filters assert created_by_condition["created_by"] == test_user_id - assert ( - len(created_by_condition) == 1 - ), "Created by condition should only have created_by field" + assert len(created_by_condition) == 1, "Created by condition should only have created_by field" # Reset mocks for Test Case 2 mock_find_many.reset_mock() @@ -187,9 +181,7 @@ async def test_list_keys_include_created_by_keys(): print(f"error: {e}") where_condition_no_created_by = mock_find_many.call_args.kwargs["where"] - print( - f"where_condition with include_created_by_keys=False: {where_condition_no_created_by}" - ) + print(f"where_condition with include_created_by_keys=False: {where_condition_no_created_by}") # Should not have OR conditions when include_created_by_keys=False and no admin_team_ids # The user condition should be merged directly into the where clause @@ -229,15 +221,11 @@ async def test_list_keys_include_created_by_keys(): created_by_condition_with_exclude = condition # Verify exclude_team_id is applied to user condition - assert ( - user_condition_with_exclude is not None - ), "User condition with exclude should be present" + assert user_condition_with_exclude is not None, "User condition with exclude should be present" assert user_condition_with_exclude["team_id"] == {"not": "excluded-team-123"} # Verify created_by condition still only has created_by filter - assert ( - created_by_condition_with_exclude is not None - ), "Created by condition with exclude should be present" + assert created_by_condition_with_exclude is not None, "Created by condition with exclude should be present" assert created_by_condition_with_exclude["created_by"] == test_user_id assert len(created_by_condition_with_exclude) == 1 @@ -251,24 +239,16 @@ async def test_key_token_handling(monkeypatch): """ mock_prisma_client = AsyncMock() mock_insert_data = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) mock_prisma_client.insert_data = mock_insert_data mock_prisma_client.db = MagicMock() mock_prisma_client.db.litellm_verificationtoken = MagicMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) mock_prisma_client.db.litellm_verificationtoken.count = AsyncMock(return_value=0) mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) from litellm.proxy._types import GenerateKeyRequest, LitellmUserRoles @@ -284,17 +264,13 @@ async def test_key_token_handling(monkeypatch): # Test key generation response = await generate_key_fn( data=GenerateKeyRequest(), - user_api_key_dict=UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1234", user_id="1234" - ), + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1234", user_id="1234"), ) # Verify token handling assert response.key != response.token, "Token should not equal key" if hasattr(response, "token_id"): - assert ( - response.token == response.token_id - ), "Token should equal token_id if token_id exists" + assert response.token == response.token_id, "Token should equal token_id if token_id exists" @pytest.mark.asyncio @@ -305,18 +281,12 @@ async def test_budget_reset_and_expires_at_first_of_month(monkeypatch): - expires is set to approximately 1 month from creation time (exact duration) """ mock_prisma_client = AsyncMock() - mock_insert_data = AsyncMock( - return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None) - ) + mock_insert_data = AsyncMock(return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None)) mock_prisma_client.insert_data = mock_insert_data mock_prisma_client.db = MagicMock() mock_prisma_client.db.litellm_verificationtoken = MagicMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) mock_prisma_client.db.litellm_verificationtoken.count = AsyncMock(return_value=0) mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None) @@ -358,15 +328,13 @@ async def test_budget_reset_and_expires_at_first_of_month(monkeypatch): # Verify budget_reset_at is set to first of next month (standardized reset time) budget_reset_at = response.get("budget_reset_at") assert budget_reset_at is not None, "budget_reset_at not found in response" - assert ( - budget_reset_at.year == expected_year - ), f"Expected year {expected_year}, got {budget_reset_at.year} for budget_reset_at" - assert ( - budget_reset_at.month == expected_month - ), f"Expected month {expected_month}, got {budget_reset_at.month} for budget_reset_at" - assert ( - budget_reset_at.day == 1 - ), f"Expected day 1, got {budget_reset_at.day} for budget_reset_at" + assert budget_reset_at.year == expected_year, ( + f"Expected year {expected_year}, got {budget_reset_at.year} for budget_reset_at" + ) + assert budget_reset_at.month == expected_month, ( + f"Expected month {expected_month}, got {budget_reset_at.month} for budget_reset_at" + ) + assert budget_reset_at.day == 1, f"Expected day 1, got {budget_reset_at.day} for budget_reset_at" # Verify expires is set to approximately 1 month from creation time (exact duration, not standardized) expires = response.get("expires") @@ -375,9 +343,9 @@ async def test_budget_reset_and_expires_at_first_of_month(monkeypatch): # Allow for some variance due to test execution time (subtract 1 second buffer for timing) expected_expires_min = now + timedelta(days=28, seconds=-1) expected_expires_max = now + timedelta(days=32) - assert ( - expected_expires_min <= expires <= expected_expires_max - ), f"Expected expires to be approximately 1 month from now, got {expires}" + assert expected_expires_min <= expires <= expected_expires_max, ( + f"Expected expires to be approximately 1 month from now, got {expires}" + ) @pytest.mark.asyncio @@ -387,18 +355,12 @@ async def test_key_expiration_exact_duration_hours(monkeypatch): Specifically tests the bug where "12h" duration would expire at midnight instead of 12 hours from creation. """ mock_prisma_client = AsyncMock() - mock_insert_data = AsyncMock( - return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None) - ) + mock_insert_data = AsyncMock(return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None)) mock_prisma_client.insert_data = mock_insert_data mock_prisma_client.db = MagicMock() mock_prisma_client.db.litellm_verificationtoken = MagicMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) mock_prisma_client.db.litellm_verificationtoken.count = AsyncMock(return_value=0) mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None) @@ -430,9 +392,9 @@ async def test_key_expiration_exact_duration_hours(monkeypatch): expected_expires_min = now + timedelta(hours=11, minutes=59) expected_expires_max = now + timedelta(hours=12, minutes=1) - assert ( - expected_expires_min <= expires <= expected_expires_max - ), f"Expected expires to be approximately 12 hours from now ({now}), got {expires}. Duration should be exact, not aligned to time boundaries." + assert expected_expires_min <= expires <= expected_expires_max, ( + f"Expected expires to be approximately 12 hours from now ({now}), got {expires}. Duration should be exact, not aligned to time boundaries." + ) # Verify it's NOT aligned to hour boundaries (e.g., not exactly at :00 minutes) # If created at 2:30 PM, it should expire at 2:30 AM, not midnight @@ -442,9 +404,9 @@ async def test_key_expiration_exact_duration_hours(monkeypatch): # More importantly, verify the duration is correct time_diff = expires - now hours_diff = time_diff.total_seconds() / 3600 - assert ( - 11.9 <= hours_diff <= 12.1 - ), f"Expected expiration to be approximately 12 hours from creation, got {hours_diff} hours" + assert 11.9 <= hours_diff <= 12.1, ( + f"Expected expiration to be approximately 12 hours from creation, got {hours_diff} hours" + ) @pytest.mark.asyncio @@ -460,14 +422,10 @@ async def test_key_generation_with_object_permission(monkeypatch): mock_prisma_client.jsonify_object = lambda data: data # type: ignore # Mock the prisma_client.db.litellm_objectpermissiontable.create call - mock_object_permission_create = AsyncMock( - return_value=MagicMock(object_permission_id="objperm123") - ) + mock_object_permission_create = AsyncMock(return_value=MagicMock(object_permission_id="objperm123")) mock_prisma_client.db = MagicMock() mock_prisma_client.db.litellm_objectpermissiontable = MagicMock() - mock_prisma_client.db.litellm_objectpermissiontable.create = ( - mock_object_permission_create - ) + mock_prisma_client.db.litellm_objectpermissiontable.create = mock_object_permission_create # Mock prisma_client.insert_data for both user and key tables async def _insert_data_side_effect(*args, **kwargs): # type: ignore @@ -500,9 +458,7 @@ async def _insert_data_side_effect(*args, **kwargs): # type: ignore ) # --- Call generate_key_fn with object_permission --- - request_data = GenerateKeyRequest( - object_permission=LiteLLM_ObjectPermissionBase(vector_stores=["my-vector"]) - ) + request_data = GenerateKeyRequest(object_permission=LiteLLM_ObjectPermissionBase(vector_stores=["my-vector"])) await generate_key_fn( data=request_data, @@ -519,9 +475,7 @@ async def _insert_data_side_effect(*args, **kwargs): # type: ignore # 2. Key insert received the generated object_permission_id key_insert_calls = [ - call.kwargs - for call in mock_prisma_client.insert_data.call_args_list - if call.kwargs.get("table_name") == "key" + call.kwargs for call in mock_prisma_client.insert_data.call_args_list if call.kwargs.get("table_name") == "key" ] assert len(key_insert_calls) == 1 assert key_insert_calls[0]["data"].get("object_permission_id") == "objperm123" @@ -585,9 +539,7 @@ async def test_generate_key_personal_non_admin_denied_for_team_scoped_fields( if "object_permission" in request_kwargs: request_kwargs = { **request_kwargs, - "object_permission": LiteLLM_ObjectPermissionBase( - **request_kwargs["object_permission"] - ), + "object_permission": LiteLLM_ObjectPermissionBase(**request_kwargs["object_permission"]), } request_data = GenerateKeyRequest(**request_kwargs) @@ -604,9 +556,7 @@ async def test_generate_key_personal_non_admin_denied_for_team_scoped_fields( ) code = getattr(exc.value, "status_code", None) or getattr(exc.value, "code", None) assert int(code) == 403 - body = str( - getattr(exc.value, "detail", None) or getattr(exc.value, "message", exc.value) - ) + body = str(getattr(exc.value, "detail", None) or getattr(exc.value, "message", exc.value)) assert expected_in_error in body mock_prisma_client.db.litellm_objectpermissiontable.create.assert_not_called() @@ -810,9 +760,7 @@ async def _insert_data_side_effect(*args, **kwargs): budget_fallbacks={"anthropic-haiku-4-5": ["gpt-5.5"]}, ) - assert json.loads(captured_key_data["budget_fallbacks"]) == { - "anthropic-haiku-4-5": ["gpt-5.5"] - } + assert json.loads(captured_key_data["budget_fallbacks"]) == {"anthropic-haiku-4-5": ["gpt-5.5"]} @pytest.mark.asyncio @@ -855,9 +803,7 @@ async def _insert_data_side_effect(*args, **kwargs): monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) monkeypatch.setattr( "litellm.proxy.management_endpoints.key_management_endpoints.validate_key_mcp_servers_against_team", - AsyncMock( - side_effect=lambda object_permission=None, **kwargs: object_permission - ), + AsyncMock(side_effect=lambda object_permission=None, **kwargs: object_permission), ) from litellm.proxy._types import ( @@ -891,9 +837,7 @@ async def _insert_data_side_effect(*args, **kwargs): assert "mcp_tool_permissions" in created_permission_data import json - assert json.loads(created_permission_data["mcp_tool_permissions"]) == { - "server_1": ["tool1", "tool2", "tool3"] - } + assert json.loads(created_permission_data["mcp_tool_permissions"]) == {"server_1": ["tool1", "tool2", "tool3"]} assert created_permission_data["mcp_servers"] == ["server_1"] @@ -937,16 +881,12 @@ async def test_key_update_object_permissions_existing_permission(monkeypatch): "vector_stores": ["old_store_1", "old_store_2"], } - mock_prisma_client.db.litellm_objectpermissiontable.find_unique = AsyncMock( - return_value=existing_object_permission - ) + mock_prisma_client.db.litellm_objectpermissiontable.find_unique = AsyncMock(return_value=existing_object_permission) # Mock upsert operation updated_permission = MagicMock() updated_permission.object_permission_id = "existing_perm_id_123" - mock_prisma_client.db.litellm_objectpermissiontable.upsert = AsyncMock( - return_value=updated_permission - ) + mock_prisma_client.db.litellm_objectpermissiontable.upsert = AsyncMock(return_value=updated_permission) # Test data with new object permission data_json = { @@ -1006,21 +946,17 @@ async def test_key_update_object_permissions_no_existing_permission(monkeypatch) ) # Mock find_unique to return None (no existing permission) - mock_prisma_client.db.litellm_objectpermissiontable.find_unique = AsyncMock( - return_value=None - ) + mock_prisma_client.db.litellm_objectpermissiontable.find_unique = AsyncMock(return_value=None) # Mock upsert to create new record new_permission = MagicMock() new_permission.object_permission_id = "new_perm_id_456" - mock_prisma_client.db.litellm_objectpermissiontable.upsert = AsyncMock( - return_value=new_permission - ) + mock_prisma_client.db.litellm_objectpermissiontable.upsert = AsyncMock(return_value=new_permission) data_json = { - "object_permission": LiteLLM_ObjectPermissionBase( - vector_stores=["brand_new_store"] - ).model_dump(exclude_unset=True, exclude_none=True), + "object_permission": LiteLLM_ObjectPermissionBase(vector_stores=["brand_new_store"]).model_dump( + exclude_unset=True, exclude_none=True + ), "user_id": "user456", } @@ -1069,21 +1005,17 @@ async def test_key_update_object_permissions_missing_permission_record(monkeypat ) # Mock find_unique to return None (permission record not found) - mock_prisma_client.db.litellm_objectpermissiontable.find_unique = AsyncMock( - return_value=None - ) + mock_prisma_client.db.litellm_objectpermissiontable.find_unique = AsyncMock(return_value=None) # Mock upsert to create new record new_permission = MagicMock() new_permission.object_permission_id = "recreated_perm_id_789" - mock_prisma_client.db.litellm_objectpermissiontable.upsert = AsyncMock( - return_value=new_permission - ) + mock_prisma_client.db.litellm_objectpermissiontable.upsert = AsyncMock(return_value=new_permission) data_json = { - "object_permission": LiteLLM_ObjectPermissionBase( - vector_stores=["recreated_store"] - ).model_dump(exclude_unset=True, exclude_none=True), + "object_permission": LiteLLM_ObjectPermissionBase(vector_stores=["recreated_store"]).model_dump( + exclude_unset=True, exclude_none=True + ), "user_id": "user789", } @@ -1208,9 +1140,7 @@ async def test_key_info_returns_object_permission(monkeypatch): mock_key_info.dict.return_value = mock_key_info.model_dump.return_value # Mock find_unique for the key lookup - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_key_info - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_key_info) # Mock object permission record mock_object_permission = MagicMock() @@ -1221,14 +1151,10 @@ async def test_key_info_returns_object_permission(monkeypatch): "vector_stores": ["vs_1", "vs_2"], "agents": ["agent_1"], } - mock_object_permission.dict.return_value = ( - mock_object_permission.model_dump.return_value - ) + mock_object_permission.dict.return_value = mock_object_permission.model_dump.return_value # Mock find_unique for object permission lookup - mock_prisma_client.db.litellm_objectpermissiontable.find_unique = AsyncMock( - return_value=mock_object_permission - ) + mock_prisma_client.db.litellm_objectpermissiontable.find_unique = AsyncMock(return_value=mock_object_permission) # Create user API key dict user_api_key_dict = UserAPIKeyAuth( @@ -1449,9 +1375,7 @@ async def test_generate_service_account_requires_team_id(): metadata={"service_account_id": "sa"}, team_id=None, ), - user_api_key_dict=UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1" - ), + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1"), litellm_changed_by=None, team_table=None, ) @@ -1470,7 +1394,6 @@ async def test_generate_service_account_works_with_team_id(): "litellm.proxy.management_endpoints.key_management_endpoints.generate_key_helper_fn" ) as mock_generate_key, ): - # Configure mocks mock_prisma.return_value = AsyncMock() mock_router.return_value = None @@ -1488,9 +1411,7 @@ async def test_generate_service_account_works_with_team_id(): metadata={"service_account_id": "sa"}, team_id="IJ", ), - user_api_key_dict=UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1" - ), + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1"), litellm_changed_by=None, team_table=None, ) @@ -1507,9 +1428,7 @@ async def test_update_service_account_requires_team_id(): @pytest.mark.asyncio async def test_update_service_account_works_with_team_id(): - data = UpdateKeyRequest( - key="sk-1", metadata={"service_account_id": "sa"}, team_id="IJ" - ) + data = UpdateKeyRequest(key="sk-1", metadata={"service_account_id": "sa"}, team_id="IJ") existing_key = LiteLLM_VerificationToken(token="hashed") await prepare_key_update_data(data=data, existing_key_row=existing_key) @@ -1696,9 +1615,7 @@ async def test_prepare_key_update_data_duration_never_expires(): # Test setting duration to "-1" (never expires) update_request = UpdateKeyRequest(key="test-token", duration="-1") - result = await prepare_key_update_data( - data=update_request, existing_key_row=existing_key - ) + result = await prepare_key_update_data(data=update_request, existing_key_row=existing_key) # Verify that expires is set to None assert result["expires"] is None @@ -1725,9 +1642,7 @@ async def test_prepare_key_update_data_duration_none_never_expires(): update_request = UpdateKeyRequest(key="test-token", duration=None) - result = await prepare_key_update_data( - data=update_request, existing_key_row=existing_key - ) + result = await prepare_key_update_data(data=update_request, existing_key_row=existing_key) assert result["expires"] is None @@ -1752,9 +1667,7 @@ async def test_prepare_key_update_data_budget_limits_clears_field(cleared_value) update_request = UpdateKeyRequest(key="test-token", budget_limits=cleared_value) - result = await prepare_key_update_data( - data=update_request, existing_key_row=existing_key - ) + result = await prepare_key_update_data(data=update_request, existing_key_row=existing_key) assert result["budget_limits"] == json.dumps(None) @@ -1781,9 +1694,7 @@ async def test_prepare_key_update_data_budget_limits_serializes_windows(): budget_limits=[{"budget_duration": "1d", "max_budget": 10.0}], ) - result = await prepare_key_update_data( - data=update_request, existing_key_row=existing_key - ) + result = await prepare_key_update_data(data=update_request, existing_key_row=existing_key) windows = json.loads(result["budget_limits"]) assert isinstance(result["budget_limits"], str) @@ -1877,9 +1788,7 @@ async def test_validate_team_id_used_in_service_account_request_requires_prisma_ ) assert exc_info.value.status_code == 400 - assert "prisma_client is required for service account keys" in str( - exc_info.value.detail - ) + assert "prisma_client is required for service account keys" in str(exc_info.value.detail) @pytest.mark.asyncio @@ -1962,16 +1871,12 @@ async def test_generate_service_account_key_endpoint_validation(): with pytest.raises(HTTPException) as exc_info: await generate_service_account_key_fn( data=GenerateKeyRequest(team_id=None), - user_api_key_dict=UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1" - ), + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1"), litellm_changed_by=None, ) assert exc_info.value.status_code == 400 - assert "team_id is required for service account keys" in str( - exc_info.value.detail - ) + assert "team_id is required for service account keys" in str(exc_info.value.detail) # Test case 2: Team doesn't exist in database with patch("litellm.proxy.proxy_server.prisma_client") as mock_prisma: @@ -1982,9 +1887,7 @@ async def test_generate_service_account_key_endpoint_validation(): with pytest.raises(HTTPException) as exc_info: await generate_service_account_key_fn( data=GenerateKeyRequest(team_id="non-existent-team"), - user_api_key_dict=UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1" - ), + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1"), litellm_changed_by=None, ) @@ -2010,25 +1913,17 @@ async def test_unblock_key_supports_both_sk_and_hashed_tokens(monkeypatch): mock_proxy_logging_obj = MagicMock() # Use a proper 64-character hex hash for testing - test_hashed_token = ( - "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" - ) + test_hashed_token = "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" # Mock the key record that will be returned from database mock_key_record = MagicMock() mock_key_record.token = test_hashed_token mock_key_record.blocked = False - mock_key_record.model_dump_json.return_value = ( - f'{{"token": "{test_hashed_token}", "blocked": false}}' - ) + mock_key_record.model_dump_json.return_value = f'{{"token": "{test_hashed_token}", "blocked": false}}' # Mock database operations - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_key_record - ) - mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=mock_key_record - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_key_record) + mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock(return_value=mock_key_record) # Mock hash_token function def mock_hash_token(token): @@ -2038,16 +1933,10 @@ def mock_hash_token(token): # Apply monkeypatch monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.hash_token", mock_hash_token) - monkeypatch.setattr( - "litellm.store_audit_logs", False - ) # Disable audit logs for simpler test + monkeypatch.setattr("litellm.store_audit_logs", False) # Disable audit logs for simpler test # Mock get_key_object and _cache_key_object async def mock_delete_cache_key_object(**kwargs): @@ -2060,9 +1949,7 @@ async def mock_delete_cache_key_object(**kwargs): # Create mock request and user auth mock_request = MagicMock() - user_api_key_dict = UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user" - ) + user_api_key_dict = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user") # Test Case 1: Using sk- prefixed token sk_token_request = BlockKeyRequest(key="sk-test123456789") @@ -2118,9 +2005,7 @@ async def test_unblock_key_invalid_key_format(monkeypatch): # Mock request and user auth mock_request = MagicMock() - user_api_key_dict = UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user" - ) + user_api_key_dict = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user") # Test with invalid key format invalid_key_request = BlockKeyRequest(key="invalid-key-format") @@ -2156,27 +2041,19 @@ async def test_block_key_nonexistent_key_returns_404(monkeypatch): mock_proxy_logging_obj = MagicMock() # find_unique returns None → key does not exist - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) def mock_hash_token(token): return "abcd1234" * 8 # 64-char hex monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.hash_token", mock_hash_token) monkeypatch.setattr("litellm.store_audit_logs", False) mock_request = MagicMock() - user_api_key_dict = UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user" - ) + user_api_key_dict = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user") data = BlockKeyRequest(key="sk-does-not-exist-key") @@ -2212,27 +2089,19 @@ async def test_unblock_key_nonexistent_key_returns_404(monkeypatch): mock_proxy_logging_obj = MagicMock() # find_unique returns None → key does not exist - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) def mock_hash_token(token): return "abcd1234" * 8 monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.hash_token", mock_hash_token) monkeypatch.setattr("litellm.store_audit_logs", False) mock_request = MagicMock() - user_api_key_dict = UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user" - ) + user_api_key_dict = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user") data = BlockKeyRequest(key="sk-does-not-exist-key") @@ -2266,24 +2135,16 @@ async def test_update_key_nonexistent_key_returns_404(monkeypatch): mock_proxy_logging_obj = MagicMock() # find_unique returns None → key does not exist - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None) monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True) mock_request = MagicMock() - user_api_key_dict = UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user" - ) + user_api_key_dict = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user") data = UpdateKeyRequest(key="sk-does-not-exist-key") @@ -2313,26 +2174,18 @@ async def test_block_key_existing_key_succeeds(monkeypatch): mock_user_api_key_cache = MagicMock() mock_proxy_logging_obj = MagicMock() - test_hashed_token = ( - "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" - ) + test_hashed_token = "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" mock_key_record = MagicMock() mock_key_record.token = test_hashed_token mock_key_record.blocked = False - mock_key_record.model_dump_json.return_value = ( - f'{{"token": "{test_hashed_token}", "blocked": false}}' - ) + mock_key_record.model_dump_json.return_value = f'{{"token": "{test_hashed_token}", "blocked": false}}' - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_key_record - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_key_record) mock_updated_record = MagicMock() mock_updated_record.token = test_hashed_token mock_updated_record.blocked = True - mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=mock_updated_record - ) + mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock(return_value=mock_updated_record) def mock_hash_token(token): if token.startswith("sk-"): @@ -2340,12 +2193,8 @@ def mock_hash_token(token): return token monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.hash_token", mock_hash_token) monkeypatch.setattr("litellm.store_audit_logs", False) @@ -2359,9 +2208,7 @@ async def mock_delete_cache_key_object(**kwargs): ) mock_request = MagicMock() - user_api_key_dict = UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user" - ) + user_api_key_dict = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin_user") data = BlockKeyRequest(key="sk-test123456789") @@ -2419,16 +2266,13 @@ async def test_validate_key_team_change_with_member_permissions(): "litellm.proxy.management_endpoints.key_management_endpoints.can_team_access_model", new_callable=AsyncMock, ): - with patch( - "litellm.proxy.management_endpoints.key_management_endpoints._get_user_in_team" - ) as mock_get_user: + with patch("litellm.proxy.management_endpoints.key_management_endpoints._get_user_in_team") as mock_get_user: with patch( "litellm.proxy.management_endpoints.key_management_endpoints._is_user_team_admin" ) as mock_is_admin: with patch( "litellm.proxy.management_endpoints.key_management_endpoints.TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint" ) as mock_has_perms: - mock_get_user.return_value = mock_member_object mock_is_admin.return_value = False mock_has_perms.return_value = True @@ -2483,9 +2327,7 @@ async def test_validate_key_team_change_skips_all_team_models_sentinel(): "litellm.proxy.management_endpoints.key_management_endpoints.can_team_access_model", new_callable=AsyncMock, ) as mock_can_access: - with patch( - "litellm.proxy.management_endpoints.key_management_endpoints._get_user_in_team" - ) as mock_get_user: + with patch("litellm.proxy.management_endpoints.key_management_endpoints._get_user_in_team") as mock_get_user: mock_get_user.return_value = MagicMock() await validate_key_team_change( @@ -2515,9 +2357,7 @@ def test_key_rotation_fields_helper(): # Simulate the rotation logic from generate_key_helper_fn if auto_rotate and rotation_interval: - key_data.update( - {"auto_rotate": auto_rotate, "rotation_interval": rotation_interval} - ) + key_data.update({"auto_rotate": auto_rotate, "rotation_interval": rotation_interval}) # Verify rotation fields are added assert key_data["auto_rotate"] == True @@ -2532,9 +2372,7 @@ def test_key_rotation_fields_helper(): # Simulate the rotation logic if auto_rotate2 and rotation_interval2: - key_data2.update( - {"auto_rotate": auto_rotate2, "rotation_interval": rotation_interval2} - ) + key_data2.update({"auto_rotate": auto_rotate2, "rotation_interval": rotation_interval2}) # Verify rotation fields are NOT added assert "auto_rotate" not in key_data2 @@ -2549,9 +2387,7 @@ def test_key_rotation_fields_helper(): # Simulate the rotation logic if auto_rotate3 and rotation_interval3: - key_data3.update( - {"auto_rotate": auto_rotate3, "rotation_interval": rotation_interval3} - ) + key_data3.update({"auto_rotate": auto_rotate3, "rotation_interval": rotation_interval3}) # Verify rotation fields are NOT added (missing interval) assert "auto_rotate" not in key_data3 @@ -2579,13 +2415,9 @@ async def test_update_key_fn_auto_rotate_enable(): ) # Test enabling auto rotation - update_request = UpdateKeyRequest( - key="test-token", auto_rotate=True, rotation_interval="30d" - ) + update_request = UpdateKeyRequest(key="test-token", auto_rotate=True, rotation_interval="30d") - result = await prepare_key_update_data( - data=update_request, existing_key_row=existing_key - ) + result = await prepare_key_update_data(data=update_request, existing_key_row=existing_key) # Verify rotation fields are included assert result["auto_rotate"] is True @@ -2615,9 +2447,7 @@ async def test_update_key_fn_auto_rotate_disable(): # Test disabling auto rotation update_request = UpdateKeyRequest(key="test-token", auto_rotate=False) - result = await prepare_key_update_data( - data=update_request, existing_key_row=existing_key - ) + result = await prepare_key_update_data(data=update_request, existing_key_row=existing_key) # Verify auto_rotate is set to False assert result["auto_rotate"] is False @@ -2631,9 +2461,7 @@ async def test_check_team_key_limits_no_existing_keys(): """ # Mock prisma client mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) # Create team table with limits team_table = LiteLLM_TeamTableCachedObj( @@ -2738,9 +2566,7 @@ async def test_check_team_key_limits_tpm_overallocation(): existing_key2.metadata = {} mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[existing_key1, existing_key2] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[existing_key1, existing_key2]) # Create team table with limits team_table = LiteLLM_TeamTableCachedObj( @@ -2771,9 +2597,8 @@ async def test_check_team_key_limits_tpm_overallocation(): ) assert exc_info.value.status_code == 400 - assert ( - "Allocated TPM limit=9000 + Key TPM limit=2000 is greater than team TPM limit=10000" - in str(exc_info.value.detail) + assert "Allocated TPM limit=9000 + Key TPM limit=2000 is greater than team TPM limit=10000" in str( + exc_info.value.detail ) @@ -2795,9 +2620,7 @@ async def test_check_team_key_limits_rpm_overallocation(): existing_key2.metadata = {} mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[existing_key1, existing_key2] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[existing_key1, existing_key2]) # Create team table with limits team_table = LiteLLM_TeamTableCachedObj( @@ -2828,9 +2651,8 @@ async def test_check_team_key_limits_rpm_overallocation(): ) assert exc_info.value.status_code == 400 - assert ( - "Allocated RPM limit=900 + Key RPM limit=200 is greater than team RPM limit=1000" - in str(exc_info.value.detail) + assert "Allocated RPM limit=900 + Key RPM limit=200 is greater than team RPM limit=1000" in str( + exc_info.value.detail ) @@ -2859,9 +2681,7 @@ async def test_check_team_key_limits_on_update_excludes_self(): other_key.metadata = {} mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[self_key, other_key] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[self_key, other_key]) team_table = LiteLLM_TeamTableCachedObj( team_id="test-team-self", @@ -2905,9 +2725,7 @@ async def test_check_team_key_limits_no_team_limits(): existing_key.rpm_limit = 500 mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[existing_key] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[existing_key]) # Create team table with no limits team_table = LiteLLM_TeamTableCachedObj( @@ -2948,9 +2766,7 @@ async def test_check_team_key_limits_no_key_limits(): existing_key.rpm_limit = 800 mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[existing_key] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[existing_key]) # Create team table with limits team_table = LiteLLM_TeamTableCachedObj( @@ -3044,9 +2860,7 @@ async def test_check_team_key_limits_exact_boundary(): existing_key.rpm_limit = 700 mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[existing_key] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[existing_key]) # Create team table with limits team_table = LiteLLM_TeamTableCachedObj( @@ -3181,9 +2995,8 @@ def test_check_team_key_model_specific_limits_rpm_overallocation(): ) assert exc_info.value.status_code == 400 - assert ( - "Allocated RPM limit=800 + Key RPM limit=300 is greater than team RPM limit=1000" - in str(exc_info.value.detail) + assert "Allocated RPM limit=800 + Key RPM limit=300 is greater than team RPM limit=1000" in str( + exc_info.value.detail ) @@ -3258,9 +3071,8 @@ def test_check_team_key_model_specific_limits_team_model_rpm_overallocation(): ) assert exc_info.value.status_code == 400 - assert ( - "Allocated RPM limit=550 + Key RPM limit=200 is greater than team RPM limit=700" - in str(exc_info.value.detail) + assert "Allocated RPM limit=550 + Key RPM limit=200 is greater than team RPM limit=700" in str( + exc_info.value.detail ) @@ -3335,9 +3147,8 @@ def test_check_team_key_model_specific_limits_team_model_tpm_overallocation(): ) assert exc_info.value.status_code == 400 - assert ( - "Allocated TPM limit=8500 + Key TPM limit=2000 is greater than team TPM limit=10000" - in str(exc_info.value.detail) + assert "Allocated TPM limit=8500 + Key TPM limit=2000 is greater than team TPM limit=10000" in str( + exc_info.value.detail ) @@ -3366,9 +3177,7 @@ async def test_generate_key_with_object_permission(): mock_prisma_client.jsonify_object = lambda x: x # Mock object permission creation - mock_object_perm_create = AsyncMock( - return_value=MagicMock(object_permission_id="objperm_key_456") - ) + mock_object_perm_create = AsyncMock(return_value=MagicMock(object_permission_id="objperm_key_456")) mock_prisma_client.db.litellm_objectpermissiontable = MagicMock() mock_prisma_client.db.litellm_objectpermissiontable.create = mock_object_perm_create @@ -3414,9 +3223,7 @@ async def test_generate_key_with_object_permission(): ), patch( "litellm.proxy.management_endpoints.key_management_endpoints.validate_key_mcp_servers_against_team", - new=AsyncMock( - side_effect=lambda object_permission=None, **kwargs: object_permission - ), + new=AsyncMock(side_effect=lambda object_permission=None, **kwargs: object_permission), ), ): # Execute @@ -3710,9 +3517,7 @@ async def test_check_org_key_limits_no_existing_keys(): """ # Mock prisma client mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) # Create organization table with limits in budget table org_table = LiteLLM_OrganizationTable( @@ -3826,9 +3631,7 @@ async def test_check_org_key_limits_tpm_overallocation(): existing_key2.metadata = {} mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[existing_key1, existing_key2] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[existing_key1, existing_key2]) # Create organization table with limits org_table = LiteLLM_OrganizationTable( @@ -3861,9 +3664,8 @@ async def test_check_org_key_limits_tpm_overallocation(): ) assert exc_info.value.status_code == 400 - assert ( - "Allocated TPM limit=18000 + Key TPM limit=3000 is greater than organization TPM limit=20000" - in str(exc_info.value.detail) + assert "Allocated TPM limit=18000 + Key TPM limit=3000 is greater than organization TPM limit=20000" in str( + exc_info.value.detail ) @@ -3885,9 +3687,7 @@ async def test_check_org_key_limits_rpm_overallocation(): existing_key2.metadata = {} mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[existing_key1, existing_key2] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[existing_key1, existing_key2]) # Create organization table with limits org_table = LiteLLM_OrganizationTable( @@ -3920,9 +3720,8 @@ async def test_check_org_key_limits_rpm_overallocation(): ) assert exc_info.value.status_code == 400 - assert ( - "Allocated RPM limit=1800 + Key RPM limit=300 is greater than organization RPM limit=2000" - in str(exc_info.value.detail) + assert "Allocated RPM limit=1800 + Key RPM limit=300 is greater than organization RPM limit=2000" in str( + exc_info.value.detail ) @@ -3939,9 +3738,7 @@ async def test_check_org_key_limits_no_org_limits(): existing_key.metadata = {} mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[existing_key] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[existing_key]) # Create organization table with no limits org_table = LiteLLM_OrganizationTable( @@ -4083,9 +3880,8 @@ def test_check_org_key_model_specific_limits_rpm_overallocation(): ) assert exc_info.value.status_code == 400 - assert ( - "Allocated RPM limit=1400 + Key RPM limit=700 is greater than organization RPM limit=2000" - in str(exc_info.value.detail) + assert "Allocated RPM limit=1400 + Key RPM limit=700 is greater than organization RPM limit=2000" in str( + exc_info.value.detail ) @@ -4162,9 +3958,8 @@ def test_check_org_key_model_specific_limits_org_model_rpm_overallocation(): ) assert exc_info.value.status_code == 400 - assert ( - "Allocated RPM limit=1100 + Key RPM limit=400 is greater than organization RPM limit=1400" - in str(exc_info.value.detail) + assert "Allocated RPM limit=1100 + Key RPM limit=400 is greater than organization RPM limit=1400" in str( + exc_info.value.detail ) @@ -4241,9 +4036,8 @@ def test_check_org_key_model_specific_limits_org_model_tpm_overallocation(): ) assert exc_info.value.status_code == 400 - assert ( - "Allocated TPM limit=17000 + Key TPM limit=4000 is greater than organization TPM limit=20000" - in str(exc_info.value.detail) + assert "Allocated TPM limit=17000 + Key TPM limit=4000 is greater than organization TPM limit=20000" in str( + exc_info.value.detail ) @@ -4305,9 +4099,7 @@ def test_transform_verification_tokens_to_deleted_records(): assert all("deleted_by_api_key" in record for record in records) assert all("litellm_changed_by" in record for record in records) assert all(record["deleted_by"] == "user-123" for record in records) - assert all( - record["deleted_by_api_key"] == user_api_key_dict.api_key for record in records - ) + assert all(record["deleted_by_api_key"] == user_api_key_dict.api_key for record in records) assert all(record["litellm_changed_by"] == "admin-user" for record in records) record1 = records[0] @@ -4349,9 +4141,7 @@ def test_transform_verification_tokens_to_deleted_records_empty_list(): async def test_save_deleted_verification_token_records(): mock_prisma_client = AsyncMock() mock_create_many = AsyncMock() - mock_prisma_client.db.litellm_deletedverificationtoken.create_many = ( - mock_create_many - ) + mock_prisma_client.db.litellm_deletedverificationtoken.create_many = mock_create_many records = [ { @@ -4368,9 +4158,7 @@ async def test_save_deleted_verification_token_records(): }, ] - await _save_deleted_verification_token_records( - records=records, prisma_client=mock_prisma_client - ) + await _save_deleted_verification_token_records(records=records, prisma_client=mock_prisma_client) mock_create_many.assert_called_once_with(data=records) @@ -4379,13 +4167,9 @@ async def test_save_deleted_verification_token_records(): async def test_save_deleted_verification_token_records_empty_list(): mock_prisma_client = AsyncMock() mock_create_many = AsyncMock() - mock_prisma_client.db.litellm_deletedverificationtoken.create_many = ( - mock_create_many - ) + mock_prisma_client.db.litellm_deletedverificationtoken.create_many = mock_create_many - await _save_deleted_verification_token_records( - records=[], prisma_client=mock_prisma_client - ) + await _save_deleted_verification_token_records(records=[], prisma_client=mock_prisma_client) mock_create_many.assert_not_called() @@ -4394,9 +4178,7 @@ async def test_save_deleted_verification_token_records_empty_list(): async def test_persist_deleted_verification_tokens(): mock_prisma_client = AsyncMock() mock_create_many = AsyncMock() - mock_prisma_client.db.litellm_deletedverificationtoken.create_many = ( - mock_create_many - ) + mock_prisma_client.db.litellm_deletedverificationtoken.create_many = mock_create_many user_api_key_dict = UserAPIKeyAuth( user_id="user-123", @@ -4502,9 +4284,7 @@ async def test_delete_verification_tokens_persists_deleted_keys(monkeypatch): mock_user_api_key_cache.delete_cache = MagicMock() mock_create_many = AsyncMock() - mock_prisma_client.db.litellm_deletedverificationtoken.create_many = ( - mock_create_many - ) + mock_prisma_client.db.litellm_deletedverificationtoken.create_many = mock_create_many def mock_hash_token(token): return token if not token.startswith("sk-") else f"hashed-{token}" @@ -5331,9 +5111,7 @@ async def test_list_keys_with_expand_user(): mock_key1.user_id = "user123" mock_key1.created_by = None # Set up model_dump() to raise AttributeError so it falls back to dict() - mock_key1.model_dump = MagicMock( - side_effect=AttributeError("model_dump not available") - ) + mock_key1.model_dump = MagicMock(side_effect=AttributeError("model_dump not available")) mock_key1.dict = MagicMock(return_value=key1_dict) key2_dict = { @@ -5347,9 +5125,7 @@ async def test_list_keys_with_expand_user(): mock_key2.user_id = "user456" mock_key2.created_by = None # Set up model_dump() to raise AttributeError so it falls back to dict() - mock_key2.model_dump = MagicMock( - side_effect=AttributeError("model_dump not available") - ) + mock_key2.model_dump = MagicMock(side_effect=AttributeError("model_dump not available")) mock_key2.dict = MagicMock(return_value=key2_dict) mock_find_many_keys = AsyncMock(return_value=[mock_key1, mock_key2]) @@ -5572,18 +5348,14 @@ async def test_list_keys_with_status_deleted(): "key_alias": "deleted_key2", } - mock_find_many_deleted = AsyncMock( - return_value=[mock_deleted_key1, mock_deleted_key2] - ) + mock_find_many_deleted = AsyncMock(return_value=[mock_deleted_key1, mock_deleted_key2]) mock_count_deleted = AsyncMock(return_value=2) # Mock regular keys table (should not be called) mock_find_many_regular = AsyncMock(return_value=[]) mock_count_regular = AsyncMock(return_value=0) - mock_prisma_client.db.litellm_deletedverificationtoken.find_many = ( - mock_find_many_deleted - ) + mock_prisma_client.db.litellm_deletedverificationtoken.find_many = mock_find_many_deleted mock_prisma_client.db.litellm_deletedverificationtoken.count = mock_count_deleted mock_prisma_client.db.litellm_verificationtoken.find_many = mock_find_many_regular mock_prisma_client.db.litellm_verificationtoken.count = mock_count_regular @@ -5717,8 +5489,7 @@ async def test_list_keys_non_admin_user_id_auto_set(): mock_list_key_helper.assert_called_once() call_kwargs = mock_list_key_helper.call_args.kwargs assert call_kwargs["user_id"] == test_user_id, ( - f"Expected user_id to be set to {test_user_id}, " - f"but got {call_kwargs.get('user_id')}" + f"Expected user_id to be set to {test_user_id}, but got {call_kwargs.get('user_id')}" ) @@ -5968,17 +5739,15 @@ def test_build_key_filter_conditions_full_visibility_team_includes_service_accou serialized = json.dumps(where) # Full-visibility team filter: no user_id restriction - assert ( - json.dumps({"team_id": {"in": [full_visibility_team]}}) in serialized - ), f"expected unrestricted team_id IN clause, got: {serialized}" + assert json.dumps({"team_id": {"in": [full_visibility_team]}}) in serialized, ( + f"expected unrestricted team_id IN clause, got: {serialized}" + ) # No service-account-only AND clause for this team (it would be redundant # and would erroneously narrow the visibility back to user_id=NULL). - sa_only_clause = json.dumps( - {"AND": [{"team_id": {"in": [full_visibility_team]}}, {"user_id": None}]} + sa_only_clause = json.dumps({"AND": [{"team_id": {"in": [full_visibility_team]}}, {"user_id": None}]}) + assert sa_only_clause not in serialized, ( + f"team in admin_team_ids must not also be filtered to user_id=NULL: {serialized}" ) - assert ( - sa_only_clause not in serialized - ), f"team in admin_team_ids must not also be filtered to user_id=NULL: {serialized}" def test_build_key_filter_conditions_member_only_team_restricts_to_service_accounts(): @@ -6007,12 +5776,8 @@ def test_build_key_filter_conditions_member_only_team_restricts_to_service_accou ) serialized = json.dumps(where) - expected = json.dumps( - {"AND": [{"team_id": {"in": [member_only_team]}}, {"user_id": None}]} - ) - assert ( - expected in serialized - ), f"member-only team must be restricted to user_id=NULL keys, got: {serialized}" + expected = json.dumps({"AND": [{"team_id": {"in": [member_only_team]}}, {"user_id": None}]}) + assert expected in serialized, f"member-only team must be restricted to user_id=NULL keys, got: {serialized}" def test_build_key_filter_conditions_agent_id_narrows_visibility(): @@ -6133,12 +5898,8 @@ async def _insert_data_side_effect(*args, **kwargs): mock_prisma_client.insert_data = AsyncMock(side_effect=_insert_data_side_effect) mock_prisma_client.db = MagicMock() mock_prisma_client.db.litellm_verificationtoken = MagicMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) mock_prisma_client.db.litellm_verificationtoken.count = AsyncMock(return_value=0) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) @@ -6174,9 +5935,7 @@ async def _insert_data_side_effect(*args, **kwargs): # Verify key insertion was called assert mock_prisma_client.insert_data.call_count >= 1 key_insert_calls = [ - call.kwargs - for call in mock_prisma_client.insert_data.call_args_list - if call.kwargs.get("table_name") == "key" + call.kwargs for call in mock_prisma_client.insert_data.call_args_list if call.kwargs.get("table_name") == "key" ] assert len(key_insert_calls) >= 1 key_data = key_insert_calls[0]["data"] @@ -6198,9 +5957,7 @@ async def _insert_data_side_effect(*args, **kwargs): # If it's still a dict, use it directly actual_settings = router_settings_value else: - raise AssertionError( - f"router_settings should be str or dict, got {type(router_settings_value)}" - ) + raise AssertionError(f"router_settings should be str or dict, got {type(router_settings_value)}") # Verify router_settings matches input (regardless of serialization state) assert actual_settings == router_settings_data @@ -6237,13 +5994,9 @@ async def test_update_key_with_router_settings(monkeypatch): "num_retries": 2, } - update_request = UpdateKeyRequest( - key="test-token-router", router_settings=router_settings_data - ) + update_request = UpdateKeyRequest(key="test-token-router", router_settings=router_settings_data) - result = await prepare_key_update_data( - data=update_request, existing_key_row=existing_key - ) + result = await prepare_key_update_data(data=update_request, existing_key_row=existing_key) # Verify router_settings is serialized to JSON string assert "router_settings" in result @@ -6285,9 +6038,7 @@ async def test_validate_max_budget(): _validate_max_budget(-10.0) assert exc_info.value.status_code == 400 - assert "max_budget must be a non-negative finite number" in str( - exc_info.value.detail - ) + assert "max_budget must be a non-negative finite number" in str(exc_info.value.detail) assert "negative" in str(exc_info.value.detail) @@ -6313,9 +6064,7 @@ async def test_get_and_validate_existing_key(): models=["gpt-4"], team_id=None, ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_key) with patch( "litellm.proxy.management_endpoints.key_management_endpoints._hash_token_if_needed", @@ -6332,9 +6081,7 @@ async def test_get_and_validate_existing_key(): ) # Test Case 2: Key not found raises ProxyException - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) with patch( "litellm.proxy.management_endpoints.key_management_endpoints._hash_token_if_needed", @@ -6396,19 +6143,13 @@ async def test_process_single_key_update(): "tags": ["production"], } - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=existing_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=existing_key) mock_updated_key_obj = MagicMock() mock_updated_key_obj.model_dump.return_value = updated_key_data - mock_prisma_client.update_data = AsyncMock( - return_value={"data": mock_updated_key_obj} - ) + mock_prisma_client.update_data = AsyncMock(return_value={"data": mock_updated_key_obj}) # Mock prepare_key_update_data - with patch( - "litellm.proxy.management_endpoints.key_management_endpoints.prepare_key_update_data" - ) as mock_prepare: + with patch("litellm.proxy.management_endpoints.key_management_endpoints.prepare_key_update_data") as mock_prepare: mock_prepare.return_value = {"max_budget": 100.0, "tags": ["production"]} # Mock TeamMemberPermissionChecks @@ -6547,18 +6288,12 @@ async def test_bulk_update_keys_success(monkeypatch): # Patch dependencies monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", mock_llm_router) # Mock helper functions - with patch( - "litellm.proxy.management_endpoints.key_management_endpoints.prepare_key_update_data" - ) as mock_prepare: + with patch("litellm.proxy.management_endpoints.key_management_endpoints.prepare_key_update_data") as mock_prepare: mock_prepare.side_effect = [ {"max_budget": 100.0, "tags": ["production"]}, {"max_budget": 200.0, "tags": ["staging"]}, @@ -6567,9 +6302,7 @@ async def test_bulk_update_keys_success(monkeypatch): with patch( "litellm.proxy.management_endpoints.key_management_endpoints.TeamMemberPermissionChecks.can_team_member_execute_key_management_endpoint" ): - with patch( - "litellm.proxy.management_endpoints.key_management_endpoints._delete_cache_key_object" - ): + with patch("litellm.proxy.management_endpoints.key_management_endpoints._delete_cache_key_object"): with patch("litellm.proxy._types.hash_token") as mock_hash: mock_hash.side_effect = ["hashed-key-1", "hashed-key-2"] @@ -6670,34 +6403,24 @@ async def test_bulk_update_keys_partial_failures(monkeypatch): ) mock_updated_key_1_obj = MagicMock() mock_updated_key_1_obj.model_dump.return_value = updated_key_1_data - mock_prisma_client.update_data = AsyncMock( - return_value={"data": mock_updated_key_1_obj} - ) + mock_prisma_client.update_data = AsyncMock(return_value={"data": mock_updated_key_1_obj}) # Mock get_data for the error handler path (used to fetch key_info on failure) mock_prisma_client.get_data = AsyncMock(return_value=None) # Patch dependencies monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", mock_llm_router) # Mock helper functions - with patch( - "litellm.proxy.management_endpoints.key_management_endpoints.prepare_key_update_data" - ) as mock_prepare: + with patch("litellm.proxy.management_endpoints.key_management_endpoints.prepare_key_update_data") as mock_prepare: mock_prepare.return_value = {"max_budget": 100.0, "tags": ["production"]} with patch( "litellm.proxy.management_endpoints.key_management_endpoints.TeamMemberPermissionChecks.can_team_member_execute_key_management_endpoint" ): - with patch( - "litellm.proxy.management_endpoints.key_management_endpoints._delete_cache_key_object" - ): + with patch("litellm.proxy.management_endpoints.key_management_endpoints._delete_cache_key_object"): with patch("litellm.proxy._types.hash_token") as mock_hash: mock_hash.return_value = "hashed-key-1" @@ -6749,10 +6472,7 @@ def _hash_for_bulk_partial(token: str) -> str: assert len(response.failed_updates) == 1 assert response.successful_updates[0].key == "test-key-1" assert response.failed_updates[0].key == "non-existent-key" - assert ( - "Key not found" - in response.failed_updates[0].failed_reason - ) + assert "Key not found" in response.failed_updates[0].failed_reason @pytest.mark.parametrize( @@ -6769,18 +6489,14 @@ def _hash_for_bulk_partial(token: str) -> str: (50.0, 100.0, 30.0, None, "reset_to (50.0) must be <= budget (30.0)"), ], ) -def test_validate_reset_spend_value_invalid( - reset_to, key_spend, key_max_budget, budget_max_budget, expected_error -): +def test_validate_reset_spend_value_invalid(reset_to, key_spend, key_max_budget, budget_max_budget, expected_error): key_in_db = LiteLLM_VerificationToken( token="test-token", user_id="test-user", spend=key_spend, max_budget=key_max_budget, litellm_budget_table=( - LiteLLM_BudgetTable( - budget_id="test-budget", max_budget=budget_max_budget - ).dict() + LiteLLM_BudgetTable(budget_id="test-budget", max_budget=budget_max_budget).dict() if budget_max_budget is not None else None ), @@ -6805,18 +6521,14 @@ def test_validate_reset_spend_value_invalid( (10.5, 50.0, 20.0, None), ], ) -def test_validate_reset_spend_value_valid( - reset_to, key_spend, key_max_budget, budget_max_budget -): +def test_validate_reset_spend_value_valid(reset_to, key_spend, key_max_budget, budget_max_budget): key_in_db = LiteLLM_VerificationToken( token="test-token", user_id="test-user", spend=key_spend, max_budget=key_max_budget, litellm_budget_table=( - LiteLLM_BudgetTable( - budget_id="test-budget", max_budget=budget_max_budget - ).dict() + LiteLLM_BudgetTable(budget_id="test-budget", max_budget=budget_max_budget).dict() if budget_max_budget is not None else None ), @@ -6880,20 +6592,12 @@ async def test_reset_key_spend_success(monkeypatch): budget_reset_at=None, ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=key_in_db - ) - mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=updated_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=key_in_db) + mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock(return_value=updated_key) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) with ( patch("litellm.proxy.proxy_server.hash_token") as mock_hash_token, @@ -6965,17 +6669,11 @@ async def test_update_key_spend_updates_counter(monkeypatch): mock_prisma_client.get_data = AsyncMock(return_value=key_in_db) mock_prisma_client.update_data = AsyncMock(return_value={"data": {"spend": 0.0}}) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=key_in_db - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=key_in_db) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None) monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True) monkeypatch.setattr("litellm.store_audit_logs", False) @@ -7044,12 +6742,8 @@ async def test_reset_key_spend_success_team_admin(monkeypatch): budget_reset_at=None, ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=key_in_db - ) - mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=updated_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=key_in_db) + mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock(return_value=updated_key) # Set up team table with user as admin team_table = LiteLLM_TeamTableCachedObj( @@ -7071,12 +6765,8 @@ async def mock_get_team_object(*args, **kwargs): return team_table monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr( "litellm.proxy.management_endpoints.key_management_endpoints.get_team_object", mock_get_team_object, @@ -7115,9 +6805,7 @@ async def mock_get_team_object(*args, **kwargs): @pytest.mark.asyncio async def test_reset_key_spend_key_not_found(monkeypatch): mock_prisma_client = MagicMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) @@ -7139,9 +6827,9 @@ async def test_reset_key_spend_key_not_found(monkeypatch): ) assert exc_info.value.status_code == 404 - assert "Key not found" in str( + assert "Key not found" in str(exc_info.value.detail) or "Key sk-test-key not found" in str( exc_info.value.detail - ) or "Key sk-test-key not found" in str(exc_info.value.detail) + ) @pytest.mark.asyncio @@ -7177,9 +6865,7 @@ async def test_reset_key_spend_validation_error(monkeypatch): litellm_budget_table=None, ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=key_in_db - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=key_in_db) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) @@ -7219,14 +6905,10 @@ async def test_reset_key_spend_authorization_failure(monkeypatch): litellm_budget_table=None, ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=key_in_db - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=key_in_db) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) with ( patch("litellm.proxy.proxy_server.hash_token") as mock_hash_token, @@ -7235,9 +6917,7 @@ async def test_reset_key_spend_authorization_failure(monkeypatch): ) as mock_check_admin, ): mock_hash_token.return_value = hashed_key - mock_check_admin.side_effect = HTTPException( - status_code=403, detail={"error": "Not authorized"} - ) + mock_check_admin.side_effect = HTTPException(status_code=403, detail={"error": "Not authorized"}) user_api_key_dict = UserAPIKeyAuth( user_role=LitellmUserRoles.INTERNAL_USER, @@ -7279,20 +6959,12 @@ async def test_reset_key_spend_hashed_key(monkeypatch): budget_reset_at=None, ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=key_in_db - ) - mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=updated_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=key_in_db) + mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock(return_value=updated_key) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) with ( patch( @@ -7355,9 +7027,7 @@ async def test_validate_key_list_check_team_admin_success(): organization_memberships=[], ) - mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( - return_value=user_info - ) + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=user_info) user_api_key_dict = UserAPIKeyAuth( user_role=LitellmUserRoles.INTERNAL_USER, @@ -7388,9 +7058,7 @@ async def test_validate_key_list_check_team_admin_fail(): organization_memberships=[], ) - mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( - return_value=user_info - ) + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=user_info) user_api_key_dict = UserAPIKeyAuth( user_role=LitellmUserRoles.INTERNAL_USER, @@ -7427,12 +7095,8 @@ async def test_validate_key_list_check_key_hash_authorized(): user_id="test-user", ) - mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( - return_value=user_info - ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=key_info - ) + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=user_info) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=key_info) user_api_key_dict = UserAPIKeyAuth( user_role=LitellmUserRoles.INTERNAL_USER, @@ -7473,13 +7137,9 @@ async def test_validate_key_list_check_key_hash_unauthorized(): user_id="other-user", ) - mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( - return_value=user_info - ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=key_info - ) - + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=user_info) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=key_info) + user_api_key_dict = UserAPIKeyAuth( user_role=LitellmUserRoles.INTERNAL_USER, user_id="test-user", @@ -7515,12 +7175,8 @@ async def test_validate_key_list_check_key_hash_not_found(): organization_memberships=[], ) - mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( - return_value=user_info - ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - side_effect=Exception("Key not found") - ) + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=user_info) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(side_effect=Exception("Key not found")) user_api_key_dict = UserAPIKeyAuth( user_role=LitellmUserRoles.INTERNAL_USER, @@ -7617,9 +7273,7 @@ async def test_key_does_not_override_explicit_budget_duration(): # The budget tier has budget_duration="7d" mock_budget_row = MagicMock() mock_budget_row.budget_duration = "7d" - mock_prisma.db.litellm_budgettable.find_unique = AsyncMock( - return_value=mock_budget_row - ) + mock_prisma.db.litellm_budgettable.find_unique = AsyncMock(return_value=mock_budget_row) mock_generate_key = AsyncMock( return_value={ @@ -7669,9 +7323,7 @@ async def test_key_does_not_override_explicit_budget_duration(): @pytest.mark.asyncio -@patch( - "litellm.proxy.management_endpoints.key_management_endpoints.rotate_mcp_server_credentials_master_key" -) +@patch("litellm.proxy.management_endpoints.key_management_endpoints.rotate_mcp_server_credentials_master_key") async def test_rotate_master_key_model_data_valid_for_prisma( mock_rotate_mcp, ): @@ -7697,15 +7349,11 @@ async def test_rotate_master_key_model_data_valid_for_prisma( mock_model = MagicMock() mock_model.model_id = "model-1" mock_model.model_name = "test-model" - mock_model.litellm_params = ( - '{"model": "openai/gpt-4", "api_key": "sk-encrypted-old"}' - ) + mock_model.litellm_params = '{"model": "openai/gpt-4", "api_key": "sk-encrypted-old"}' mock_model.model_info = '{"id": "model-1"}' mock_model.created_by = "admin" mock_model.updated_by = "admin" - mock_prisma_client.db.litellm_proxymodeltable.find_many = AsyncMock( - return_value=[mock_model] - ) + mock_prisma_client.db.litellm_proxymodeltable.find_many = AsyncMock(return_value=[mock_model]) # Mock transaction context manager mock_tx = AsyncMock() @@ -7723,9 +7371,7 @@ async def test_rotate_master_key_model_data_valid_for_prisma( mock_prisma_client.db.litellm_config.find_many = AsyncMock(return_value=[]) # Mock credentials table — no credentials - mock_prisma_client.db.litellm_credentialstable.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_credentialstable.find_many = AsyncMock(return_value=[]) # Mock MCP rotation mock_rotate_mcp.return_value = None @@ -7771,22 +7417,18 @@ async def test_rotate_master_key_model_data_valid_for_prisma( model_data = created_models[0] # Verify timestamps are NOT present (Prisma @default(now()) should apply) - assert ( - "created_at" not in model_data - ), "created_at should be excluded so Prisma @default(now()) applies" - assert ( - "updated_at" not in model_data - ), "updated_at should be excluded so Prisma @default(now()) applies" + assert "created_at" not in model_data, "created_at should be excluded so Prisma @default(now()) applies" + assert "updated_at" not in model_data, "updated_at should be excluded so Prisma @default(now()) applies" # Verify litellm_params and model_info are prisma.Json wrappers, NOT JSON strings import prisma - assert isinstance( - model_data["litellm_params"], prisma.Json - ), f"litellm_params should be prisma.Json for create_many(), got {type(model_data['litellm_params'])}" - assert isinstance( - model_data["model_info"], prisma.Json - ), f"model_info should be prisma.Json for create_many(), got {type(model_data['model_info'])}" + assert isinstance(model_data["litellm_params"], prisma.Json), ( + f"litellm_params should be prisma.Json for create_many(), got {type(model_data['litellm_params'])}" + ) + assert isinstance(model_data["model_info"], prisma.Json), ( + f"model_info should be prisma.Json for create_many(), got {type(model_data['model_info'])}" + ) # Verify delete_many was called inside the transaction (before create_many) mock_tx.litellm_proxymodeltable.delete_many.assert_called_once() @@ -7804,24 +7446,16 @@ async def test_default_key_generate_params_duration(monkeypatch): mock_prisma_client = AsyncMock() mock_insert_data = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) mock_prisma_client.insert_data = mock_insert_data mock_prisma_client.db = MagicMock() mock_prisma_client.db.litellm_verificationtoken = MagicMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) mock_prisma_client.db.litellm_verificationtoken.count = AsyncMock(return_value=0) mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) @@ -7860,24 +7494,16 @@ async def test_default_key_generate_params_object_permission_applied_when_absent mock_prisma_client = AsyncMock() mock_insert_data = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) mock_prisma_client.insert_data = mock_insert_data mock_prisma_client.db = MagicMock() mock_prisma_client.db.litellm_verificationtoken = MagicMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) mock_prisma_client.db.litellm_verificationtoken.count = AsyncMock(return_value=0) mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) mock_prisma_client.db.litellm_objectpermissiontable = MagicMock() mock_prisma_client.db.litellm_objectpermissiontable.create = AsyncMock( @@ -7887,9 +7513,7 @@ async def test_default_key_generate_params_object_permission_applied_when_absent monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) original_value = litellm.default_key_generate_params - litellm.default_key_generate_params = { - "object_permission": {"vector_stores": ["default-vs"]} - } + litellm.default_key_generate_params = {"object_permission": {"vector_stores": ["default-vs"]}} try: request = GenerateKeyRequest() # No object_permission specified @@ -7904,11 +7528,7 @@ async def test_default_key_generate_params_object_permission_applied_when_absent team_table=None, ) - created_data = ( - mock_prisma_client.db.litellm_objectpermissiontable.create.call_args.kwargs[ - "data" - ] - ) + created_data = mock_prisma_client.db.litellm_objectpermissiontable.create.call_args.kwargs["data"] assert created_data["vector_stores"] == ["default-vs"] finally: litellm.default_key_generate_params = original_value @@ -7927,24 +7547,16 @@ async def test_default_key_generate_params_object_permission_merges_partial( mock_prisma_client = AsyncMock() mock_insert_data = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) mock_prisma_client.insert_data = mock_insert_data mock_prisma_client.db = MagicMock() mock_prisma_client.db.litellm_verificationtoken = MagicMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) mock_prisma_client.db.litellm_verificationtoken.count = AsyncMock(return_value=0) mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) mock_prisma_client.db.litellm_objectpermissiontable = MagicMock() mock_prisma_client.db.litellm_objectpermissiontable.create = AsyncMock( @@ -7954,14 +7566,10 @@ async def test_default_key_generate_params_object_permission_merges_partial( monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) original_value = litellm.default_key_generate_params - litellm.default_key_generate_params = { - "object_permission": {"vector_stores": ["default-vs"]} - } + litellm.default_key_generate_params = {"object_permission": {"vector_stores": ["default-vs"]}} try: - request = GenerateKeyRequest( - object_permission=LiteLLM_ObjectPermissionBase(agents=["agent-1"]) - ) + request = GenerateKeyRequest(object_permission=LiteLLM_ObjectPermissionBase(agents=["agent-1"])) await _common_key_generation_helper( data=request, user_api_key_dict=UserAPIKeyAuth( @@ -7973,11 +7581,7 @@ async def test_default_key_generate_params_object_permission_merges_partial( team_table=None, ) - created_data = ( - mock_prisma_client.db.litellm_objectpermissiontable.create.call_args.kwargs[ - "data" - ] - ) + created_data = mock_prisma_client.db.litellm_objectpermissiontable.create.call_args.kwargs["data"] assert created_data["agents"] == ["agent-1"] assert created_data["vector_stores"] == ["default-vs"] finally: @@ -7996,24 +7600,16 @@ async def test_default_key_generate_params_object_permission_does_not_override_e mock_prisma_client = AsyncMock() mock_insert_data = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) mock_prisma_client.insert_data = mock_insert_data mock_prisma_client.db = MagicMock() mock_prisma_client.db.litellm_verificationtoken = MagicMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) mock_prisma_client.db.litellm_verificationtoken.count = AsyncMock(return_value=0) mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) mock_prisma_client.db.litellm_objectpermissiontable = MagicMock() mock_prisma_client.db.litellm_objectpermissiontable.create = AsyncMock( @@ -8023,16 +7619,10 @@ async def test_default_key_generate_params_object_permission_does_not_override_e monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) original_value = litellm.default_key_generate_params - litellm.default_key_generate_params = { - "object_permission": {"vector_stores": ["default-vs"]} - } + litellm.default_key_generate_params = {"object_permission": {"vector_stores": ["default-vs"]}} try: - request = GenerateKeyRequest( - object_permission=LiteLLM_ObjectPermissionBase( - vector_stores=["explicit-vs"] - ) - ) + request = GenerateKeyRequest(object_permission=LiteLLM_ObjectPermissionBase(vector_stores=["explicit-vs"])) await _common_key_generation_helper( data=request, user_api_key_dict=UserAPIKeyAuth( @@ -8044,11 +7634,7 @@ async def test_default_key_generate_params_object_permission_does_not_override_e team_table=None, ) - created_data = ( - mock_prisma_client.db.litellm_objectpermissiontable.create.call_args.kwargs[ - "data" - ] - ) + created_data = mock_prisma_client.db.litellm_objectpermissiontable.create.call_args.kwargs["data"] assert created_data["vector_stores"] == ["explicit-vs"] finally: litellm.default_key_generate_params = original_value @@ -8068,24 +7654,16 @@ async def test_default_key_generate_params_object_permission_not_rejected_for_no mock_prisma_client = AsyncMock() mock_insert_data = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) mock_prisma_client.insert_data = mock_insert_data mock_prisma_client.db = MagicMock() mock_prisma_client.db.litellm_verificationtoken = MagicMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=None - ) - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) mock_prisma_client.db.litellm_verificationtoken.count = AsyncMock(return_value=0) mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=MagicMock( - token="hashed_token_123", litellm_budget_table=None, object_permission=None - ) + return_value=MagicMock(token="hashed_token_123", litellm_budget_table=None, object_permission=None) ) mock_prisma_client.db.litellm_objectpermissiontable = MagicMock() mock_prisma_client.db.litellm_objectpermissiontable.create = AsyncMock( @@ -8099,9 +7677,7 @@ async def test_default_key_generate_params_object_permission_not_rejected_for_no ) original_value = litellm.default_key_generate_params - litellm.default_key_generate_params = { - "object_permission": {"vector_stores": ["default-vs"]} - } + litellm.default_key_generate_params = {"object_permission": {"vector_stores": ["default-vs"]}} try: request = GenerateKeyRequest(user_id="alice") # No object_permission specified @@ -8117,11 +7693,7 @@ async def test_default_key_generate_params_object_permission_not_rejected_for_no ) assert response is not None - created_data = ( - mock_prisma_client.db.litellm_objectpermissiontable.create.call_args.kwargs[ - "data" - ] - ) + created_data = mock_prisma_client.db.litellm_objectpermissiontable.create.call_args.kwargs["data"] assert created_data["vector_stores"] == ["default-vs"] finally: litellm.default_key_generate_params = original_value @@ -8219,9 +7791,7 @@ async def test_build_key_filter_admin_sees_all_team_keys(): assert admin_cond["team_id"]["in"] == admin_team_ids # member-only condition should only include team-B (team-A is covered by admin) - assert ( - service_account_cond is not None - ), "Service account condition should be present" + assert service_account_cond is not None, "Service account condition should be present" and_parts = service_account_cond["AND"] assert {"team_id": {"in": ["team-B"]}} in and_parts assert {"user_id": None} in and_parts @@ -8409,10 +7979,7 @@ async def test_build_key_filter_admin_all_member_overlap(): if "AND" in cond: and_parts = cond["AND"] if {"user_id": None} in and_parts: - pytest.fail( - "Should not have member-only service account condition " - "when user is admin of all teams" - ) + pytest.fail("Should not have member-only service account condition when user is admin of all teams") @pytest.mark.asyncio @@ -8556,9 +8123,7 @@ async def test_build_key_filter_team_id_scoped(): # not buried inside an OR branch (which was the bug). assert "AND" in where, f"Expected top-level AND, got: {where}" outer_and = where["AND"] - assert { - "team_id": "team-A" - } in outer_and, ( + assert {"team_id": "team-A"} in outer_and, ( f"Expected {{'team_id': 'team-A'}} as a direct AND condition, got: {outer_and}" ) @@ -8687,9 +8252,7 @@ async def test_get_member_team_ids(): mock_team_a.model_dump.return_value = { "team_id": "team-A", "team_alias": "Team A", - "members_with_roles": [ - {"user_id": user_id, "role": "admin", "user_email": None} - ], + "members_with_roles": [{"user_id": user_id, "role": "admin", "user_email": None}], "max_budget": None, "budget_duration": None, "budget_reset_at": None, @@ -8703,9 +8266,7 @@ async def test_get_member_team_ids(): mock_team_b.model_dump.return_value = { "team_id": "team-B", "team_alias": "Team B", - "members_with_roles": [ - {"user_id": user_id, "role": "user", "user_email": None} - ], + "members_with_roles": [{"user_id": user_id, "role": "user", "user_email": None}], "max_budget": None, "budget_duration": None, "budget_reset_at": None, @@ -8719,9 +8280,7 @@ async def test_get_member_team_ids(): mock_team_c.model_dump.return_value = { "team_id": "team-C", "team_alias": "Team C", - "members_with_roles": [ - {"user_id": "other-user", "role": "admin", "user_email": None} - ], + "members_with_roles": [{"user_id": "other-user", "role": "admin", "user_email": None}], "max_budget": None, "budget_duration": None, "budget_reset_at": None, @@ -8731,9 +8290,7 @@ async def test_get_member_team_ids(): "blocked": False, } - mock_prisma_client.db.litellm_teamtable.find_many = AsyncMock( - return_value=[mock_team_a, mock_team_b, mock_team_c] - ) + mock_prisma_client.db.litellm_teamtable.find_many = AsyncMock(return_value=[mock_team_a, mock_team_b, mock_team_c]) result = await get_member_team_ids( complete_user_info=user_info, @@ -8793,9 +8350,9 @@ async def test_generate_key_helper_fn_agent_id(): # insert_data is called as insert_data(data=key_data, ...) call_kwargs = mock_insert.call_args.kwargs key_data = call_kwargs.get("data", {}) - assert ( - key_data.get("agent_id") == "test-agent-456" - ), f"Expected agent_id='test-agent-456' in key_data, got: {key_data.get('agent_id')}" + assert key_data.get("agent_id") == "test-agent-456", ( + f"Expected agent_id='test-agent-456' in key_data, got: {key_data.get('agent_id')}" + ) def _make_admin_key_dict() -> UserAPIKeyAuth: @@ -8925,9 +8482,7 @@ async def test_key_aliases_internal_user_scoped_to_own_keys_and_teams(): # Mock user table lookup to return teams mock_user_row = MagicMock() mock_user_row.teams = ["team-abc", "team-xyz"] - mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock( - return_value=mock_user_row - ) + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=mock_user_row) mock_prisma_client.db.query_raw = AsyncMock( side_effect=[ @@ -9064,9 +8619,7 @@ async def test_check_org_key_limits_on_update_within_bounds(): a key's TPM/RPM limits within organization bounds. """ mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[]) org_table = LiteLLM_OrganizationTable( organization_id="test-org-update-1", @@ -9118,9 +8671,7 @@ async def test_check_org_key_limits_on_update_overallocation(): existing_key.metadata = {} mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[existing_key] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[existing_key]) org_table = LiteLLM_OrganizationTable( organization_id="test-org-update-2", @@ -9179,9 +8730,7 @@ async def test_check_org_key_limits_on_update_excludes_self(): other_key.metadata = {} mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[self_key, other_key] - ) + mock_prisma_client.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[self_key, other_key]) org_table = LiteLLM_OrganizationTable( organization_id="test-org-self", @@ -9245,15 +8794,11 @@ def _check_throughput_changed(data: UpdateKeyRequest) -> bool: assert _check_throughput_changed(data_with_org) is True # Updating tpm_limit_type — limit type change triggers check - data_with_tpm_type = UpdateKeyRequest( - key="sk-test-key", tpm_limit_type="guaranteed_throughput" - ) + data_with_tpm_type = UpdateKeyRequest(key="sk-test-key", tpm_limit_type="guaranteed_throughput") assert _check_throughput_changed(data_with_tpm_type) is True # Updating rpm_limit_type — limit type change triggers check - data_with_rpm_type = UpdateKeyRequest( - key="sk-test-key", rpm_limit_type="guaranteed_throughput" - ) + data_with_rpm_type = UpdateKeyRequest(key="sk-test-key", rpm_limit_type="guaranteed_throughput") assert _check_throughput_changed(data_with_rpm_type) is True @@ -9283,24 +8828,16 @@ def _setup_block_unblock_mocks(monkeypatch, mock_key_team_id=None): mock_user_api_key_cache = MagicMock() mock_proxy_logging_obj = MagicMock() - test_hashed_token = ( - "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" - ) + test_hashed_token = "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" mock_key_record = MagicMock() mock_key_record.token = test_hashed_token mock_key_record.blocked = False mock_key_record.team_id = mock_key_team_id - mock_key_record.model_dump_json.return_value = ( - f'{{"token": "{test_hashed_token}", "blocked": false}}' - ) + mock_key_record.model_dump_json.return_value = f'{{"token": "{test_hashed_token}", "blocked": false}}' - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_key_record - ) - mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=mock_key_record - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_key_record) + mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock(return_value=mock_key_record) mock_key_object = MagicMock() mock_key_object.blocked = True @@ -9311,12 +8848,8 @@ def mock_hash_token(token): return token monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.hash_token", mock_hash_token) monkeypatch.setattr("litellm.store_audit_logs", False) @@ -9461,9 +8994,7 @@ async def test_update_key_max_budget_rejected_for_internal_user(monkeypatch): mock_user_api_key_cache = AsyncMock() mock_proxy_logging_obj = MagicMock() - test_hashed_token = ( - "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" - ) + test_hashed_token = "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" # Mock existing key row mock_existing_key = MagicMock() @@ -9481,17 +9012,11 @@ async def test_update_key_max_budget_rejected_for_internal_user(monkeypatch): } mock_prisma_client.get_data = AsyncMock(return_value=mock_existing_key) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_existing_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_existing_key) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None) monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True) @@ -9526,9 +9051,7 @@ async def test_update_key_non_budget_fields_allowed_for_internal_user(monkeypatc mock_user_api_key_cache = AsyncMock() mock_proxy_logging_obj = MagicMock() - test_hashed_token = ( - "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" - ) + test_hashed_token = "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" # Mock existing key row mock_existing_key = MagicMock() @@ -9553,20 +9076,12 @@ async def test_update_key_non_budget_fields_allowed_for_internal_user(monkeypatc mock_updated_key.key_alias = "my-alias" mock_prisma_client.get_data = AsyncMock(return_value=mock_existing_key) - mock_prisma_client.update_data = AsyncMock( - return_value={"data": {"key_alias": "my-alias"}} - ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_existing_key - ) + mock_prisma_client.update_data = AsyncMock(return_value={"data": {"key_alias": "my-alias"}}) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_existing_key) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None) monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True) monkeypatch.setattr("litellm.store_audit_logs", False) @@ -9652,18 +9167,14 @@ async def test_update_key_non_budget_rejects_cross_user_modification(monkeypatch } mock_prisma_client.get_data = AsyncMock(return_value=mock_existing_key) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_existing_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_existing_key) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", AsyncMock()) monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()) monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None) monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True) - monkeypatch.setattr( - "litellm.proxy.proxy_server.hash_token", lambda t: test_hashed_token - ) + monkeypatch.setattr("litellm.proxy.proxy_server.hash_token", lambda t: test_hashed_token) mock_request = MagicMock() mock_request.query_params = {} @@ -9678,9 +9189,7 @@ async def test_update_key_non_budget_rejects_cross_user_modification(monkeypatch with pytest.raises(ProxyException) as exc: await update_key_fn( request=mock_request, - data=UpdateKeyRequest( - key=test_hashed_token, key_alias="pwned", blocked=True - ), + data=UpdateKeyRequest(key=test_hashed_token, key_alias="pwned", blocked=True), user_api_key_dict=attacker, litellm_changed_by=None, ) @@ -9716,9 +9225,7 @@ async def test_update_key_creator_reassigned_key_blocked(monkeypatch): } mock_prisma_client.get_data = AsyncMock(return_value=mock_existing_key) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_existing_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_existing_key) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", AsyncMock()) @@ -9726,9 +9233,7 @@ async def test_update_key_creator_reassigned_key_blocked(monkeypatch): monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None) monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True) monkeypatch.setattr("litellm.store_audit_logs", False) - monkeypatch.setattr( - "litellm.proxy.proxy_server.hash_token", lambda t: test_hashed_token - ) + monkeypatch.setattr("litellm.proxy.proxy_server.hash_token", lambda t: test_hashed_token) demoted_admin = UserAPIKeyAuth( user_role=LitellmUserRoles.INTERNAL_USER, @@ -9802,12 +9307,8 @@ async def test_update_key_team_member_with_permission_can_update_non_budget( mock_prisma_client = AsyncMock() mock_prisma_client.get_data = AsyncMock(return_value=mock_existing_key) - mock_prisma_client.update_data = AsyncMock( - return_value={"data": {"key_alias": "renamed-by-member"}} - ) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_existing_key - ) + mock_prisma_client.update_data = AsyncMock(return_value={"data": {"key_alias": "renamed-by-member"}}) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_existing_key) async def mock_get_team_object(*args, **kwargs): return team_table @@ -9844,9 +9345,7 @@ async def mock_delete_cache_key_object(**kwargs): "litellm.proxy.management_endpoints.key_management_endpoints.get_ui_settings_cached", AsyncMock(return_value={}), ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.hash_token", lambda t: test_hashed_token - ) + monkeypatch.setattr("litellm.proxy.proxy_server.hash_token", lambda t: test_hashed_token) monkeypatch.setattr( "litellm.proxy.management_endpoints.key_management_endpoints._get_user_email_by_id", AsyncMock(return_value=None), @@ -9919,9 +9418,7 @@ async def test_update_key_team_member_cannot_change_budget(monkeypatch): mock_prisma_client = AsyncMock() mock_prisma_client.get_data = AsyncMock(return_value=mock_existing_key) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_existing_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_existing_key) async def mock_get_team_object(*args, **kwargs): return team_table @@ -9939,9 +9436,7 @@ async def mock_get_team_object(*args, **kwargs): monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()) monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None) monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True) - monkeypatch.setattr( - "litellm.proxy.proxy_server.hash_token", lambda t: test_hashed_token - ) + monkeypatch.setattr("litellm.proxy.proxy_server.hash_token", lambda t: test_hashed_token) mock_request = MagicMock() mock_request.query_params = {} @@ -10302,14 +9797,10 @@ async def test_assigned_user_blocked_from_any_edit(self): data = UpdateKeyRequest(key="sk-test", key_alias="hacked") # user_id matches caller so _is_allowed_to_make_key_request passes, # but created_by != caller so our creator check requires admin. - existing = self._make_existing_key( - user_id="assigned-user", created_by="admin-456" - ) + existing = self._make_existing_key(user_id="assigned-user", created_by="admin-456") auth = self._make_auth(user_id="assigned-user") - mock_check = AsyncMock( - side_effect=HTTPException(status_code=403, detail="Not authorized") - ) + mock_check = AsyncMock(side_effect=HTTPException(status_code=403, detail="Not authorized")) with patch( "litellm.proxy.management_endpoints.key_management_endpoints._check_key_admin_access", mock_check, @@ -10341,9 +9832,7 @@ async def test_creator_cannot_reset_own_spend_to_stale_value(self): data = UpdateKeyRequest(key="sk-test", spend=0.0) auth = self._make_auth(user_id="creator-123") - mock_check = AsyncMock( - side_effect=HTTPException(status_code=403, detail="Not authorized") - ) + mock_check = AsyncMock(side_effect=HTTPException(status_code=403, detail="Not authorized")) with patch( "litellm.proxy.management_endpoints.key_management_endpoints._check_key_admin_access", mock_check, @@ -10363,14 +9852,10 @@ async def test_creator_cannot_reset_own_spend_to_stale_value(self): @pytest.mark.asyncio async def test_assigned_user_blocked_from_model_escalation(self): data = UpdateKeyRequest(key="sk-test", models=["gpt-4", "claude-opus"]) - existing = self._make_existing_key( - user_id="assigned-user", created_by="admin-456" - ) + existing = self._make_existing_key(user_id="assigned-user", created_by="admin-456") auth = self._make_auth(user_id="assigned-user") - mock_check = AsyncMock( - side_effect=HTTPException(status_code=403, detail="Not authorized") - ) + mock_check = AsyncMock(side_effect=HTTPException(status_code=403, detail="Not authorized")) with patch( "litellm.proxy.management_endpoints.key_management_endpoints._check_key_admin_access", mock_check, @@ -10390,9 +9875,7 @@ async def test_assigned_user_blocked_from_model_escalation(self): @pytest.mark.asyncio async def test_creator_can_edit_own_key(self): """Key creator can update any non-budget field without admin.""" - data = UpdateKeyRequest( - key="sk-test", models=["gpt-4"], rpm_limit=500, key_alias="my-key" - ) + data = UpdateKeyRequest(key="sk-test", models=["gpt-4"], rpm_limit=500, key_alias="my-key") existing = self._make_existing_key(created_by="creator-123") auth = self._make_auth(user_id="creator-123") @@ -10420,9 +9903,7 @@ async def test_creator_cannot_change_own_budget(self): existing.max_budget = 10.0 auth = self._make_auth(user_id="creator-123") - mock_check = AsyncMock( - side_effect=HTTPException(status_code=403, detail="Not authorized") - ) + mock_check = AsyncMock(side_effect=HTTPException(status_code=403, detail="Not authorized")) with patch( "litellm.proxy.management_endpoints.key_management_endpoints._check_key_admin_access", mock_check, @@ -10447,9 +9928,7 @@ async def test_creator_cannot_clear_own_budget_limits(self, cleared_value): existing = self._make_existing_key(created_by="creator-123") auth = self._make_auth(user_id="creator-123") - mock_check = AsyncMock( - side_effect=HTTPException(status_code=403, detail="Not authorized") - ) + mock_check = AsyncMock(side_effect=HTTPException(status_code=403, detail="Not authorized")) with patch( "litellm.proxy.management_endpoints.key_management_endpoints._check_key_admin_access", mock_check, @@ -10553,9 +10032,7 @@ def existing_key_with_legacy_alias(self): ) @pytest.mark.asyncio - async def test_update_key_unchanged_legacy_alias_passes( - self, mock_prisma, existing_key_with_legacy_alias - ): + async def test_update_key_unchanged_legacy_alias_passes(self, mock_prisma, existing_key_with_legacy_alias): """ Updating a key without changing its key_alias should skip format validation — even if the alias wouldn't pass current rules. @@ -10569,9 +10046,7 @@ async def test_update_key_unchanged_legacy_alias_passes( from litellm.proxy.management_endpoints import key_management_endpoints as mod original_pattern = mod._KEY_ALIAS_PATTERN - mod._KEY_ALIAS_PATTERN = re.compile( - r"^[a-zA-Z0-9][a-zA-Z0-9_\-/\.]{0,253}[a-zA-Z0-9]$" - ) + mod._KEY_ALIAS_PATTERN = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_\-/\.]{0,253}[a-zA-Z0-9]$") try: # Confirm the alias WOULD fail validation directly with pytest.raises(ProxyException): @@ -10592,9 +10067,7 @@ async def test_update_key_unchanged_legacy_alias_passes( mod._KEY_ALIAS_PATTERN = original_pattern @pytest.mark.asyncio - async def test_update_key_changed_alias_still_validated( - self, mock_prisma, existing_key_with_legacy_alias - ): + async def test_update_key_changed_alias_still_validated(self, mock_prisma, existing_key_with_legacy_alias): """ When the alias IS being changed, validation should still run. """ @@ -10611,9 +10084,7 @@ async def test_update_key_changed_alias_still_validated( _validate_key_alias_format(new_alias) @pytest.mark.asyncio - async def test_update_key_changed_to_valid_alias_passes( - self, mock_prisma, existing_key_with_legacy_alias - ): + async def test_update_key_changed_to_valid_alias_passes(self, mock_prisma, existing_key_with_legacy_alias): """ Changing the alias to a new valid value should pass validation. """ @@ -10676,9 +10147,7 @@ def test_enforce_upperbound_fills_defaults_on_generate(): original = litellm.upperbound_key_generate_params try: - litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams( - tpm_limit=1000, rpm_limit=100 - ) + litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams(tpm_limit=1000, rpm_limit=100) data = GenerateKeyRequest() # tpm_limit=None, rpm_limit=None _enforce_upperbound_key_params(data, fill_defaults=True) assert data.tpm_limit == 1000 @@ -10696,9 +10165,7 @@ def test_enforce_upperbound_skips_none_on_update(): original = litellm.upperbound_key_generate_params try: - litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams( - tpm_limit=1000, rpm_limit=100 - ) + litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams(tpm_limit=1000, rpm_limit=100) data = UpdateKeyRequest(key="sk-test") # tpm_limit=None, rpm_limit=None _enforce_upperbound_key_params(data, fill_defaults=False) assert data.tpm_limit is None # should NOT be filled @@ -10740,9 +10207,7 @@ def test_enforce_upperbound_allows_within_limit_on_update(): litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams( tpm_limit=1000, rpm_limit=100, max_budget=10.0 ) - data = UpdateKeyRequest( - key="sk-test", tpm_limit=500, rpm_limit=50, max_budget=5.0 - ) + data = UpdateKeyRequest(key="sk-test", tpm_limit=500, rpm_limit=50, max_budget=5.0) _enforce_upperbound_key_params(data, fill_defaults=False) # Should not raise assert data.tpm_limit == 500 @@ -10761,9 +10226,7 @@ def test_enforce_upperbound_duration_over_limit(): original = litellm.upperbound_key_generate_params try: - litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams( - duration="7d" - ) + litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams(duration="7d") data = UpdateKeyRequest(key="sk-test", duration="30d") with pytest.raises(HTTPException) as exc_info: _enforce_upperbound_key_params(data, fill_defaults=False) @@ -10811,9 +10274,7 @@ def __iter__(self): } ) ) - mock_prisma_client.db.litellm_verificationtoken.create = AsyncMock( - return_value=None - ) + mock_prisma_client.db.litellm_verificationtoken.create = AsyncMock(return_value=None) mock_prisma_client.jsonify_object = MagicMock(side_effect=lambda data: data) return mock_prisma_client @@ -10850,9 +10311,7 @@ async def test_execute_virtual_key_regeneration_rejects_over_limit_duration(): original = litellm.upperbound_key_generate_params try: - litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams( - duration="1h" - ) + litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams(duration="1h") existing_key = _make_regenerate_existing_key() data = RegenerateKeyRequest(duration="2h") user_api_key_dict = _make_regenerate_user_api_key_dict() @@ -10906,9 +10365,7 @@ async def test_execute_virtual_key_regeneration_allows_within_limit_duration(): original = litellm.upperbound_key_generate_params try: - litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams( - duration="1h" - ) + litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams(duration="1h") existing_key = _make_regenerate_existing_key() data = RegenerateKeyRequest(duration="30m") user_api_key_dict = _make_regenerate_user_api_key_dict() @@ -10962,9 +10419,7 @@ async def test_execute_virtual_key_regeneration_rejects_over_limit_max_budget(): original = litellm.upperbound_key_generate_params try: - litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams( - max_budget=10.0 - ) + litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams(max_budget=10.0) existing_key = _make_regenerate_existing_key() data = RegenerateKeyRequest(max_budget=500.0) user_api_key_dict = _make_regenerate_user_api_key_dict() @@ -11018,9 +10473,7 @@ async def test_execute_virtual_key_regeneration_skips_none_values(): original = litellm.upperbound_key_generate_params try: - litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams( - duration="1h" - ) + litellm.upperbound_key_generate_params = LiteLLM_UpperboundKeyGenerateParams(duration="1h") existing_key = _make_regenerate_existing_key() data = RegenerateKeyRequest() # all fields None user_api_key_dict = _make_regenerate_user_api_key_dict() @@ -11564,9 +11017,7 @@ async def test_process_single_key_update_cache_invalidation_with_token_hash(): ) mock_prisma_client = AsyncMock() - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=existing_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=existing_key) mock_updated = MagicMock() mock_updated.model_dump.return_value = {"max_budget": 100.0} mock_prisma_client.update_data = AsyncMock(return_value={"data": mock_updated}) @@ -11715,13 +11166,9 @@ def __iter__(self): return iter(self._data.items()) mock_prisma_client.db.litellm_verificationtoken.update = AsyncMock( - return_value=DictLikeResult( - {"token": "new-hashed-token", "key_name": "sk-...ab12", "user_id": "user-1"} - ) - ) - mock_prisma_client.db.litellm_verificationtoken.create = AsyncMock( - return_value=None + return_value=DictLikeResult({"token": "new-hashed-token", "key_name": "sk-...ab12", "user_id": "user-1"}) ) + mock_prisma_client.db.litellm_verificationtoken.create = AsyncMock(return_value=None) mock_prisma_client.jsonify_object = MagicMock(side_effect=lambda data: data) mock_user_api_key_cache = MagicMock() @@ -11794,15 +11241,11 @@ def _make_team_key(token: str, team_id: str = "team-abc") -> LiteLLM_Verificatio def _admin() -> UserAPIKeyAuth: - return UserAPIKeyAuth( - user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin" - ) + return UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-admin", user_id="admin") def _internal_user() -> UserAPIKeyAuth: - return UserAPIKeyAuth( - user_role=LitellmUserRoles.INTERNAL_USER, api_key="sk-iu", user_id="iu" - ) + return UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, api_key="sk-iu", user_id="iu") def _updated(payload): @@ -11821,9 +11264,7 @@ def _setup_team_keys_mocks( ): """Set up mocks for bulk_update_team_keys; returns mock_prisma.""" mock_prisma = AsyncMock() - mock_prisma.db.litellm_verificationtoken.find_many = AsyncMock( - return_value=[] if find_many is None else find_many - ) + mock_prisma.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[] if find_many is None else find_many) if find_unique is not None: mock_prisma.db.litellm_verificationtoken.find_unique = find_unique if update_data is not None: @@ -11838,9 +11279,7 @@ def _setup_team_keys_mocks( AsyncMock(return_value={"max_budget": 50.0}), ) monkeypatch.setattr(f"{_BULK_PKG}._delete_cache_key_object", AsyncMock()) - monkeypatch.setattr( - f"{_BULK_PKG}.KeyManagementEventHooks.async_key_updated_hook", AsyncMock() - ) + monkeypatch.setattr(f"{_BULK_PKG}.KeyManagementEventHooks.async_key_updated_hook", AsyncMock()) monkeypatch.setattr(f"{_BULK_PKG}.get_team_object", AsyncMock(return_value=None)) monkeypatch.setattr(f"{_BULK_PKG}._check_team_key_limits", AsyncMock()) monkeypatch.setattr( @@ -11858,9 +11297,7 @@ async def _call_as_admin(data): bulk_update_team_keys, ) - return await bulk_update_team_keys( - data=data, user_api_key_dict=_admin(), litellm_changed_by=None - ) + return await bulk_update_team_keys(data=data, user_api_key_dict=_admin(), litellm_changed_by=None) # ---- happy paths ---------------------------------------------------------- @@ -11879,9 +11316,7 @@ async def test_bulk_update_team_keys_success_with_key_ids(monkeypatch): monkeypatch, find_many=keys, find_unique=find_unique, - update_data=AsyncMock( - side_effect=[{"data": _updated({"max_budget": 50.0})}] * 2 - ), + update_data=AsyncMock(side_effect=[{"data": _updated({"max_budget": 50.0})}] * 2), ) response = await _call_as_admin( @@ -11913,9 +11348,7 @@ async def test_bulk_update_team_keys_success_all_keys_in_team(monkeypatch): monkeypatch, find_many=keys, find_unique=find_unique, - update_data=AsyncMock( - side_effect=[{"data": _updated({"max_budget": 50.0})}] * 3 - ), + update_data=AsyncMock(side_effect=[{"data": _updated({"max_budget": 50.0})}] * 3), ) response = await _call_as_admin( @@ -11933,11 +11366,7 @@ async def test_bulk_update_team_keys_success_all_keys_in_team(monkeypatch): blocked_or, expires_or = where["AND"][0]["OR"], where["AND"][1]["OR"] assert {"blocked": False} in blocked_or and {"blocked": None} in blocked_or assert {"expires": None} in expires_or - assert any( - "gt" in c.get("expires", {}) - for c in expires_or - if isinstance(c.get("expires"), dict) - ) + assert any("gt" in c.get("expires", {}) for c in expires_or if isinstance(c.get("expires"), dict)) find_unique.assert_not_called() @@ -12131,9 +11560,7 @@ def test_bulk_update_team_keys_request_validation(): valid = KeyUpdateFields(max_budget=10) with pytest.raises(ValidationError): - BulkUpdateTeamKeysRequest( - team_id="t", key_ids=["k"], all_keys_in_team=True, update_fields=valid - ) + BulkUpdateTeamKeysRequest(team_id="t", key_ids=["k"], all_keys_in_team=True, update_fields=valid) with pytest.raises(ValidationError): BulkUpdateTeamKeysRequest(team_id="t", update_fields=valid) @@ -12152,9 +11579,7 @@ async def test_bulk_update_team_keys_hashes_raw_sk_key_ids(monkeypatch): raw_sk = "sk-rawkey1234567890" hashed = hash_token(raw_sk) - row = LiteLLM_VerificationToken( - token=hashed, user_id="u", models=[], team_id="team-abc", max_budget=None - ) + row = LiteLLM_VerificationToken(token=hashed, user_id="u", models=[], team_id="team-abc", max_budget=None) mock = _setup_team_keys_mocks( monkeypatch, find_many=[row], @@ -12217,9 +11642,7 @@ async def test_bulk_update_team_keys_auth_check_runs_when_no_keys_match(monkeypa @pytest.mark.asyncio -async def test_bulk_update_team_keys_does_not_log_raw_sk_token_on_failure( - monkeypatch, caplog -): +async def test_bulk_update_team_keys_does_not_log_raw_sk_token_on_failure(monkeypatch, caplog): """Regression: per-key failure must not log the raw sk-... (ERROR-level logs persist).""" import logging @@ -12278,9 +11701,7 @@ async def fake_process(*, update_key_request, **kw): BulkUpdateTeamKeysRequest( team_id="team-abc", key_ids=["tok-a"], - update_fields=KeyUpdateFields( - tpm_limit=10_000, tpm_limit_type="guaranteed_throughput" - ), + update_fields=KeyUpdateFields(tpm_limit=10_000, tpm_limit_type="guaranteed_throughput"), ) ) assert captured[0].team_id == "team-abc" @@ -12338,9 +11759,7 @@ async def test_bulk_update_team_keys_blocks_metadata_allowed_passthrough_routes( request = BulkUpdateTeamKeysRequest( team_id="team-abc", all_keys_in_team=True, - update_fields=KeyUpdateFields( - metadata={"allowed_passthrough_routes": ["/admin/*"]} - ), + update_fields=KeyUpdateFields(metadata={"allowed_passthrough_routes": ["/admin/*"]}), ) with pytest.raises(HTTPException) as exc: @@ -12417,9 +11836,7 @@ def _patch_regenerate_side_effects(): "same_user_id_allowed", ], ) -async def test_regenerate_user_id_rebind_guard( - incoming_user_id, expected_status, expected_substring -): +async def test_regenerate_user_id_rebind_guard(incoming_user_id, expected_status, expected_substring): from litellm.proxy._types import RegenerateKeyRequest from litellm.proxy.management_endpoints.key_management_endpoints import ( _execute_virtual_key_regeneration, @@ -12845,9 +12262,9 @@ async def test_ghsa_q775_ui_session_token_team_key_exempt_from_budget_ceiling(): ) except (HTTPException, ProxyException) as err: msg = str(getattr(err, "detail", "")) + str(getattr(err, "message", "")) - assert ( - "cannot exceed" not in msg.lower() - ), "UI/CLI session token creating a team key must be exempt from the ceiling" + assert "cannot exceed" not in msg.lower(), ( + "UI/CLI session token creating a team key must be exempt from the ceiling" + ) @pytest.mark.asyncio @@ -12954,9 +12371,7 @@ async def test_prepare_key_update_data_budget_duration_null_clears_fields(): update_request = UpdateKeyRequest(key="test-token", budget_duration=None) - result = await prepare_key_update_data( - data=update_request, existing_key_row=existing_key - ) + result = await prepare_key_update_data(data=update_request, existing_key_row=existing_key) assert "budget_duration" in result assert result["budget_duration"] is None @@ -12981,9 +12396,7 @@ async def test_prepare_key_update_data_budget_duration_not_sent_excluded(): update_request = UpdateKeyRequest(key="test-token", models=["gpt-4"]) - result = await prepare_key_update_data( - data=update_request, existing_key_row=existing_key - ) + result = await prepare_key_update_data(data=update_request, existing_key_row=existing_key) assert "budget_duration" not in result assert "budget_reset_at" not in result @@ -13006,9 +12419,7 @@ async def test_prepare_key_update_data_budget_duration_valid_sets_reset(): update_request = UpdateKeyRequest(key="test-token", budget_duration="30d") - result = await prepare_key_update_data( - data=update_request, existing_key_row=existing_key - ) + result = await prepare_key_update_data(data=update_request, existing_key_row=existing_key) assert result["budget_duration"] == "30d" assert result["budget_reset_at"] is not None @@ -13034,9 +12445,7 @@ async def test_info_key_fn_includes_model_max_budget_usage(monkeypatch): monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) mock_user_api_key_cache = AsyncMock() mock_user_api_key_cache.async_get_cache = AsyncMock(return_value=0.23) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) mock_key_info = MagicMock(spec=LiteLLM_VerificationToken) mock_key_info.token = test_key_token @@ -13054,9 +12463,7 @@ async def test_info_key_fn_includes_model_max_budget_usage(monkeypatch): } mock_key_info.dict.return_value = mock_key_info.model_dump.return_value - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_key_info - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_key_info) mock_prisma_client.db.query_raw = AsyncMock() user_api_key_dict = UserAPIKeyAuth( @@ -13091,9 +12498,7 @@ async def test_info_key_fn_no_model_max_budget_skips_usage(monkeypatch): monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) mock_user_api_key_cache = AsyncMock() mock_user_api_key_cache.async_get_cache = AsyncMock() - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) mock_key_info = MagicMock(spec=LiteLLM_VerificationToken) mock_key_info.token = test_key_token @@ -13111,9 +12516,7 @@ async def test_info_key_fn_no_model_max_budget_skips_usage(monkeypatch): } mock_key_info.dict.return_value = mock_key_info.model_dump.return_value - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_key_info - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_key_info) mock_prisma_client.db.query_raw = AsyncMock() user_api_key_dict = UserAPIKeyAuth( @@ -13148,9 +12551,7 @@ async def test_info_key_fn_v2_includes_model_max_budget_usage(monkeypatch): monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) mock_user_api_key_cache = AsyncMock() mock_user_api_key_cache.async_get_cache = AsyncMock(return_value=0.55) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) mock_key = MagicMock(spec=LiteLLM_VerificationToken) mock_key.token = test_key_token @@ -13207,9 +12608,7 @@ async def test_info_key_fn_budget_table_fallback(monkeypatch): monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) mock_user_api_key_cache = AsyncMock() mock_user_api_key_cache.async_get_cache = AsyncMock(return_value=1.20) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) mock_key_info = MagicMock(spec=LiteLLM_VerificationToken) mock_key_info.token = test_key_token @@ -13232,9 +12631,7 @@ async def test_info_key_fn_budget_table_fallback(monkeypatch): } mock_key_info.dict.return_value = mock_key_info.model_dump.return_value - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_key_info - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_key_info) mock_prisma_client.db.query_raw = AsyncMock() user_api_key_dict = UserAPIKeyAuth( @@ -13275,9 +12672,7 @@ async def test_info_key_fn_v2_budget_table_fallback(monkeypatch): monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) mock_user_api_key_cache = AsyncMock() mock_user_api_key_cache.async_get_cache = AsyncMock(return_value=2.50) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) mock_key = MagicMock(spec=LiteLLM_VerificationToken) mock_key.token = test_key_token @@ -13335,9 +12730,7 @@ async def test_info_key_fn_provider_prefix_spend_fallback(monkeypatch): monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) mock_user_api_key_cache = AsyncMock() mock_user_api_key_cache.async_get_cache = AsyncMock(side_effect=[None, 0.75]) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) mock_key_info = MagicMock(spec=LiteLLM_VerificationToken) mock_key_info.token = test_key_token @@ -13355,9 +12748,7 @@ async def test_info_key_fn_provider_prefix_spend_fallback(monkeypatch): } mock_key_info.dict.return_value = mock_key_info.model_dump.return_value - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_key_info - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_key_info) mock_prisma_client.db.query_raw = AsyncMock() user_api_key_dict = UserAPIKeyAuth( @@ -13408,9 +12799,7 @@ async def test_build_model_max_budget_usage_reads_current_cache_window(): ) assert result["gpt-4o"]["current_spend"] == 0.30 - mock_user_api_key_cache.async_get_cache.assert_awaited_once_with( - key="virtual_key_spend:some-hash:gpt-4o:30d" - ) + mock_user_api_key_cache.async_get_cache.assert_awaited_once_with(key="virtual_key_spend:some-hash:gpt-4o:30d") @pytest.mark.asyncio @@ -13470,9 +12859,7 @@ async def test_build_model_max_budget_usage_unparseable_duration_skipped(): result = await _build_model_max_budget_usage( api_key_hash="some-hash", - model_max_budget={ - "gpt-4o": {"budget_limit": 1.0, "budget_duration": "not-valid"} - }, + model_max_budget={"gpt-4o": {"budget_limit": 1.0, "budget_duration": "not-valid"}}, user_api_key_cache=mock_user_api_key_cache, ) assert result == {} @@ -13545,9 +12932,7 @@ async def _list_keys_capture_helper_kwargs(user_api_key_dict, **list_kwargs): teams=[], organization_memberships=[], ) - helper = AsyncMock( - return_value={"keys": [], "total_count": 0, "current_page": 1, "total_pages": 0} - ) + helper = AsyncMock(return_value={"keys": [], "total_count": 0, "current_page": 1, "total_pages": 0}) with patch("litellm.proxy.proxy_server.prisma_client", AsyncMock()): with patch( "litellm.proxy.management_endpoints.key_management_endpoints.validate_key_list_check", @@ -13572,9 +12957,7 @@ async def test_list_keys_admin_exact_by_default(): no substring_matching flag must get exact matching, so an integration scoping to one user with an admin key never receives other users' keys.""" admin = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, user_id="admin-1") - kwargs = await _list_keys_capture_helper_kwargs( - admin, user_id="alice", substring_matching=False - ) + kwargs = await _list_keys_capture_helper_kwargs(admin, user_id="alice", substring_matching=False) assert kwargs["user_id"] == "alice" assert kwargs["use_substring_matching"] is False @@ -13583,9 +12966,7 @@ async def test_list_keys_admin_exact_by_default(): async def test_list_keys_admin_substring_opt_in(): """An admin may opt back into substring matching (dashboard search).""" admin = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, user_id="admin-1") - kwargs = await _list_keys_capture_helper_kwargs( - admin, user_id="alice", substring_matching=True - ) + kwargs = await _list_keys_capture_helper_kwargs(admin, user_id="alice", substring_matching=True) assert kwargs["use_substring_matching"] is True @@ -13594,9 +12975,7 @@ async def test_list_keys_non_admin_cannot_opt_into_substring(): """substring_matching is admin-only: a non-admin requesting it still gets exact matching, scoped to their own user_id.""" user = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="alice") - kwargs = await _list_keys_capture_helper_kwargs( - user, user_id=None, substring_matching=True - ) + kwargs = await _list_keys_capture_helper_kwargs(user, user_id=None, substring_matching=True) assert kwargs["use_substring_matching"] is False assert kwargs["user_id"] == "alice" @@ -13801,9 +13180,7 @@ async def test_budget_limits_admin_unrestricted(monkeypatch): @pytest.mark.asyncio @pytest.mark.parametrize("non_finite", [float("nan"), float("inf"), float("-inf")]) -async def test_budget_limits_window_non_finite_rejected_for_non_admin( - monkeypatch, non_finite -): +async def test_budget_limits_window_non_finite_rejected_for_non_admin(monkeypatch, non_finite): """A non-admin caller submitting a non-finite `budget_limits` window gets 400. The finite-number invariant applies before role / ceiling checks.""" @@ -13833,9 +13210,7 @@ async def test_budget_limits_window_non_finite_rejected_for_non_admin( @pytest.mark.asyncio @pytest.mark.parametrize("non_finite", [float("nan"), float("inf"), float("-inf")]) -async def test_budget_limits_window_non_finite_rejected_for_admin( - monkeypatch, non_finite -): +async def test_budget_limits_window_non_finite_rejected_for_admin(monkeypatch, non_finite): """The finite-number invariant applies to every caller including proxy admin.""" monkeypatch.setattr( @@ -14367,6 +13742,10 @@ async def test_enforce_email_prefix_on_key_alias_leaves_alias_unchanged_when_fla "litellm.proxy.management_endpoints.key_management_endpoints.get_ui_settings_cached", AsyncMock(return_value={}), ) + monkeypatch.setattr( + "litellm.proxy.proxy_server.general_settings", + {}, + ) result = await _enforce_email_prefix_on_key_alias( prisma_client=None, @@ -14378,6 +13757,29 @@ async def test_enforce_email_prefix_on_key_alias_leaves_alias_unchanged_when_fla assert result == "TestKey Name" +@pytest.mark.asyncio +async def test_enforce_email_prefix_on_key_alias_uses_general_settings_fallback( + monkeypatch, +): + monkeypatch.setattr( + "litellm.proxy.management_endpoints.key_management_endpoints.get_ui_settings_cached", + AsyncMock(return_value={}), + ) + monkeypatch.setattr( + "litellm.proxy.proxy_server.general_settings", + {"enforce_email_prefix_on_key_alias": True}, + ) + + result = await _enforce_email_prefix_on_key_alias( + prisma_client=None, + key_alias="TestKey Name", + owner_email="owner.user@example.com", + fallback_email="caller@example.com", + ) + + assert result == "owner.user-test-key-name" + + @pytest.mark.asyncio async def test_enforce_email_prefix_on_key_alias_uses_caller_fallback(monkeypatch): monkeypatch.setattr( @@ -14529,9 +13931,7 @@ async def test_prepare_key_update_data_applies_email_prefix_on_changed_user_id( prisma_client = MagicMock() mock_user_repository = MagicMock() mock_user_repository.find_by_id = AsyncMock( - return_value=LiteLLM_UserTable( - user_id="new-user", user_email="new.owner@example.com" - ) + return_value=LiteLLM_UserTable(user_id="new-user", user_email="new.owner@example.com") ) monkeypatch.setattr( "litellm.proxy.management_endpoints.key_management_endpoints.UserRepository", @@ -14543,13 +13943,9 @@ async def test_prepare_key_update_data_applies_email_prefix_on_changed_user_id( ) result = await prepare_key_update_data( - data=UpdateKeyRequest( - key="hashed-token", user_id="new-user", key_alias="TestKey Name" - ), + data=UpdateKeyRequest(key="hashed-token", user_id="new-user", key_alias="TestKey Name"), existing_key_row=existing_key, - user_api_key_dict=UserAPIKeyAuth( - user_id="caller", user_email="caller@example.com" - ), + user_api_key_dict=UserAPIKeyAuth(user_id="caller", user_email="caller@example.com"), prisma_client=prisma_client, ) @@ -14582,9 +13978,7 @@ async def test_prepare_key_update_data_falls_back_to_caller_email(monkeypatch): result = await prepare_key_update_data( data=UpdateKeyRequest(key="hashed-token", key_alias="Renamed Alias"), existing_key_row=existing_key, - user_api_key_dict=UserAPIKeyAuth( - user_id="caller", user_email="caller.user@example.com" - ), + user_api_key_dict=UserAPIKeyAuth(user_id="caller", user_email="caller.user@example.com"), prisma_client=None, ) @@ -14619,9 +14013,7 @@ async def test_prepare_key_update_data_without_alias_leaves_key_alias_unchanged( result = await prepare_key_update_data( data=UpdateKeyRequest(key="hashed-token", models=["gpt-4"]), existing_key_row=existing_key, - user_api_key_dict=UserAPIKeyAuth( - user_id="caller", user_email="caller.user@example.com" - ), + user_api_key_dict=UserAPIKeyAuth(user_id="caller", user_email="caller.user@example.com"), prisma_client=None, ) @@ -14638,9 +14030,7 @@ async def test_update_key_non_budget_fields_apply_email_prefix(monkeypatch): mock_user_api_key_cache = AsyncMock() mock_proxy_logging_obj = MagicMock() - test_hashed_token = ( - "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" - ) + test_hashed_token = "a1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" mock_existing_key = MagicMock() mock_existing_key.token = test_hashed_token @@ -14668,17 +14058,11 @@ async def mock_update_data(*args, **kwargs): mock_prisma_client.get_data = AsyncMock(return_value=mock_existing_key) mock_prisma_client.update_data = AsyncMock(side_effect=mock_update_data) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_existing_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_existing_key) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) - monkeypatch.setattr( - "litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache - ) - monkeypatch.setattr( - "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj - ) + monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", mock_user_api_key_cache) + monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj) monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None) monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True) monkeypatch.setattr("litellm.store_audit_logs", False) @@ -14727,9 +14111,7 @@ async def test_update_key_non_budget_fields_reject_empty_normalized_alias(monkey ) mock_prisma_client = AsyncMock() - test_hashed_token = ( - "c1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" - ) + test_hashed_token = "c1b2c3d4e5f6789012345678901234567890123456789012345678901234abcd" mock_existing_key = MagicMock() mock_existing_key.token = test_hashed_token @@ -14750,9 +14132,7 @@ async def test_update_key_non_budget_fields_reject_empty_normalized_alias(monkey } mock_prisma_client.get_data = AsyncMock(return_value=mock_existing_key) - mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock( - return_value=mock_existing_key - ) + mock_prisma_client.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=mock_existing_key) monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", AsyncMock()) From dbe4042aba5559f29b95aa020f116bf6854e8b25 Mon Sep 17 00:00:00 2001 From: Fabrizio Siciliano Date: Tue, 7 Jul 2026 12:23:21 +0200 Subject: [PATCH 2/2] feat(proxy): add enforce_email_prefix_on_key_alias to true --- terraform/litellm/aws/config/prod/terraform.tfvars | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/terraform/litellm/aws/config/prod/terraform.tfvars b/terraform/litellm/aws/config/prod/terraform.tfvars index 7e7ac8226a7..b7c6d94ab76 100644 --- a/terraform/litellm/aws/config/prod/terraform.tfvars +++ b/terraform/litellm/aws/config/prod/terraform.tfvars @@ -51,11 +51,12 @@ proxy_config = { }, ] general_settings = { - master_key = "os.environ/LITELLM_MASTER_KEY" - database_url = "os.environ/DATABASE_URL" - alerting = ["email", "slack_budget_alerts", "slack"] - alert_types = ["budget_alerts", "spend_reports"] - alerting_threshold = 300 + master_key = "os.environ/LITELLM_MASTER_KEY" + database_url = "os.environ/DATABASE_URL" + enforce_email_prefix_on_key_alias = true + alerting = ["email", "slack_budget_alerts", "slack"] + alert_types = ["budget_alerts", "spend_reports"] + alerting_threshold = 300 alerting_args = { daily_report_frequency = 43200 # 12 hours in seconds report_check_interval = 3600 # 1 hour in seconds