From 58ecf91d4b41688c2ad3b98273a656115ad7cc43 Mon Sep 17 00:00:00 2001 From: aoright <102943475+aoright@users.noreply.github.com> Date: Mon, 28 Sep 2026 14:41:35 +0800 Subject: [PATCH] style(proxy): format model_management_endpoints with ruff --- .../model_management_endpoints.py | 4 +- .../test_model_management_endpoints.py | 1263 ++++++++++------- 2 files changed, 763 insertions(+), 504 deletions(-) diff --git a/litellm/proxy/management_endpoints/model_management_endpoints.py b/litellm/proxy/management_endpoints/model_management_endpoints.py index f17406c054d..877581633eb 100644 --- a/litellm/proxy/management_endpoints/model_management_endpoints.py +++ b/litellm/proxy/management_endpoints/model_management_endpoints.py @@ -401,9 +401,7 @@ def _raise_on_unroutable_model_deployment( incoming_params: GenericLiteLLMParams | None, existing_params: GenericLiteLLMParams | None = None, ) -> None: - violation = _validate_routable_model_deployment( - incoming_params=incoming_params, existing_params=existing_params - ) + violation = _validate_routable_model_deployment(incoming_params=incoming_params, existing_params=existing_params) if violation is None: return raise ProxyException( diff --git a/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py index c9d236bd454..a30dbf23b36 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py @@ -63,11 +63,7 @@ class MockPrismaClient: return LiteLLM_TeamTable( team_id=where["team_id"], team_alias="test_team", - members_with_roles=[ - Member( - user_id="test_user", role="admin" if self.user_admin else "user" - ) - ], + members_with_roles=[Member(user_id="test_user", role="admin" if self.user_admin else "user")], ) return None @@ -81,10 +77,7 @@ class MockPrismaClient: # Support model_name startswith filter (used by _get_team_deployments) if where and "model_name" in where: model_name_filter = where["model_name"] - if ( - isinstance(model_name_filter, dict) - and "startswith" in model_name_filter - ): + if isinstance(model_name_filter, dict) and "startswith" in model_name_filter: prefix = model_name_filter["startswith"] results = [d for d in results if d.model_name.startswith(prefix)] @@ -129,13 +122,9 @@ class MockProxyConfig: class TestModelManagementAuthChecks: def setup_method(self): """Setup test cases""" - self.admin_user = UserAPIKeyAuth( - user_id="test_admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + self.admin_user = UserAPIKeyAuth(user_id="test_admin", user_role=LitellmUserRoles.PROXY_ADMIN) - self.normal_user = UserAPIKeyAuth( - user_id="test_user", user_role=LitellmUserRoles.INTERNAL_USER - ) + self.normal_user = UserAPIKeyAuth(user_id="test_user", user_role=LitellmUserRoles.INTERNAL_USER) self.team_admin_user = UserAPIKeyAuth( user_id="test_user", @@ -154,7 +143,7 @@ class TestModelManagementAuthChecks: @pytest.mark.asyncio async def test_can_user_make_team_model_call_non_premium_fails(self): """Test that non-premium users cannot make team model calls""" - with pytest.raises(Exception, match='You must be a LiteLLM Enterprise user to use this feature\\.') as exc_info: + with pytest.raises(Exception, match="You must be a LiteLLM Enterprise user to use this feature\\.") as exc_info: ModelManagementAuthChecks.can_user_make_team_model_call( team_id="test_team", user_api_key_dict=self.admin_user, @@ -168,9 +157,7 @@ class TestModelManagementAuthChecks: team_obj = LiteLLM_TeamTable( team_id="test_team", team_alias="test_team", - members_with_roles=[ - Member(user_id=self.team_admin_user.user_id, role="admin") - ], + members_with_roles=[Member(user_id=self.team_admin_user.user_id, role="admin")], ) result = ModelManagementAuthChecks.can_user_make_team_model_call( @@ -209,7 +196,7 @@ class TestModelManagementAuthChecks: ) prisma_client = MockPrismaClient(team_exists=True) - with pytest.raises(Exception, match='You must be a LiteLLM Enterprise user to use this feature\\.') as exc_info: + with pytest.raises(Exception, match="You must be a LiteLLM Enterprise user to use this feature\\.") as exc_info: await ModelManagementAuthChecks.allow_team_model_action( model_params=model_params, user_api_key_dict=self.admin_user, @@ -318,9 +305,7 @@ class TestModelManagementAuthChecks: ModelManagementAuthChecks.can_user_attach_credential( litellm_params=litellm_params(litellm_credential_name=None), user_api_key_dict=self.team_admin_user, - existing_litellm_params=LiteLLM_Params( - model="test_model", litellm_credential_name="shared-credential" - ), + existing_litellm_params=LiteLLM_Params(model="test_model", litellm_credential_name="shared-credential"), null_detaches=True, ) @@ -333,9 +318,7 @@ class TestModelManagementAuthChecks: result = ModelManagementAuthChecks.can_user_attach_credential( litellm_params=litellm_params(litellm_credential_name=None), user_api_key_dict=self.admin_user, - existing_litellm_params=LiteLLM_Params( - model="test_model", litellm_credential_name="shared-credential" - ), + existing_litellm_params=LiteLLM_Params(model="test_model", litellm_credential_name="shared-credential"), null_detaches=True, ) @@ -359,9 +342,7 @@ class TestModelManagementAuthChecks: result = ModelManagementAuthChecks.can_user_attach_credential( litellm_params=litellm_params(litellm_credential_name=None), user_api_key_dict=self.team_admin_user, - existing_litellm_params=LiteLLM_Params( - model="test_model", litellm_credential_name="shared-credential" - ), + existing_litellm_params=LiteLLM_Params(model="test_model", litellm_credential_name="shared-credential"), ) assert result is True @@ -386,9 +367,15 @@ class TestModelManagementAuthChecks: mock_prisma = MagicMock() with ( - patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.prisma_client", mock_prisma + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam patch( # test-quality-ok: prior auth check needs a live DB; only the credential check is under test "litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call", new=AsyncMock(return_value=None), @@ -423,10 +410,18 @@ class TestModelManagementAuthChecks: model_info={"id": model_id}, ) with ( - patch("litellm.proxy.proxy_server.prisma_client", MagicMock()), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.llm_router", MagicMock()), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.prisma_client", MagicMock() + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", MagicMock() + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam patch( # test-quality-ok: stubs the DB row fetch; only the credential check is under test "litellm.proxy.management_endpoints.model_management_endpoints.get_db_model", new=AsyncMock(return_value=db_model), @@ -516,9 +511,15 @@ class TestModelManagementAuthChecks: mock_prisma = MagicMock() with ( - patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.prisma_client", mock_prisma + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam patch( # test-quality-ok: prior auth check needs a live DB; only the session tag check is under test "litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call", new=AsyncMock(return_value=None), @@ -559,10 +560,18 @@ class TestModelManagementAuthChecks: model_info={"id": model_id, "team_id": "test_team"}, ) with ( - patch("litellm.proxy.proxy_server.prisma_client", MagicMock()), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.llm_router", MagicMock()), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.prisma_client", MagicMock() + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", MagicMock() + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam patch( # test-quality-ok: stubs the DB row fetch; only the session tag check is under test "litellm.proxy.management_endpoints.model_management_endpoints.get_db_model", new=AsyncMock(return_value=db_model), @@ -611,10 +620,18 @@ class TestModelManagementAuthChecks: mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=existing_row) mock_prisma.db.litellm_proxymodeltable.update = AsyncMock() with ( - patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.llm_router", None), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.prisma_client", mock_prisma + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", None + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam patch( # test-quality-ok: prior auth check needs a live DB; only the session tag check is under test "litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call", new=AsyncMock(return_value=None), @@ -704,29 +721,21 @@ class TestDeleteTeamModelAlias: mock_prisma.db = MockPrismaWrapper(model_aliases_list) # Call the function - await delete_team_model_alias( - public_model_name="public_model_1", prisma_client=mock_prisma - ) + await delete_team_model_alias(public_model_name="public_model_1", prisma_client=mock_prisma) # Verify results mock_db = mock_prisma.db.litellm_modeltable - assert ( - len(mock_db.update_calls) == 2 - ) # Should have 2 update calls since public_model_1 appears twice + assert len(mock_db.update_calls) == 2 # Should have 2 update calls since public_model_1 appears twice # Verify first update first_update = mock_db.update_calls[0] assert first_update["where"] == {"id": 1} - assert json.loads(first_update["data"]["model_aliases"]) == { - "alias2": "public_model_2" - } + assert json.loads(first_update["data"]["model_aliases"]) == {"alias2": "public_model_2"} # Verify second update second_update = mock_db.update_calls[1] assert second_update["where"] == {"id": 2} - assert json.loads(second_update["data"]["model_aliases"]) == { - "alias3": "public_model_3" - } + assert json.loads(second_update["data"]["model_aliases"]) == {"alias3": "public_model_3"} @pytest.mark.asyncio async def test_delete_team_model_alias_no_matches(self): @@ -762,9 +771,7 @@ class TestDeleteTeamModelAlias: mock_prisma.db = MockPrismaWrapper(model_aliases_list) # Call the function with non-existent model - await delete_team_model_alias( - public_model_name="non_existent_model", prisma_client=mock_prisma - ) + await delete_team_model_alias(public_model_name="non_existent_model", prisma_client=mock_prisma) # Verify no updates were made mock_db = mock_prisma.db.litellm_modeltable @@ -1374,18 +1381,12 @@ class TestUpdateModel: updated_row.model_dump_json.return_value = "{}" mock_prisma = MagicMock() - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=existing_row - ) - mock_prisma.db.litellm_proxymodeltable.update = AsyncMock( - return_value=updated_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=existing_row) + mock_prisma.db.litellm_proxymodeltable.update = AsyncMock(return_value=updated_row) mock_router = MagicMock() mock_router.get_model_ids.return_value = [model_id] - admin_user = UserAPIKeyAuth( - user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + admin_user = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) with ( patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), @@ -1402,9 +1403,7 @@ class TestUpdateModel: ), patch( # test-quality-ok: [TQ008] isolate persistence from router reload implementation "litellm.proxy.management_endpoints.model_management_endpoints.clear_cache", - new=AsyncMock( - return_value=ReconcileOutcome(still_desired=None, live_after=None) - ), + new=AsyncMock(return_value=ReconcileOutcome(still_desired=None, live_after=None)), ) as mock_clear_cache, ): await update_model( @@ -1460,16 +1459,16 @@ class TestUpdateModel: ): await update_model( model_params=updateDeployment( - litellm_params=updateLiteLLMParams( - model="openai/gpt-4o-mini", litellm_credential_name=None - ), + litellm_params=updateLiteLLMParams(model="openai/gpt-4o-mini", litellm_credential_name=None), model_info=ModelInfo(id=model_id), ), user_api_key_dict=team_admin, ) mock_prisma.db.litellm_proxymodeltable.update.assert_awaited_once() - persisted = json.loads(mock_prisma.db.litellm_proxymodeltable.update.await_args.kwargs["data"]["litellm_params"]) + persisted = json.loads( + mock_prisma.db.litellm_proxymodeltable.update.await_args.kwargs["data"]["litellm_params"] + ) assert persisted["litellm_credential_name"] == "shared-credential" @@ -1503,9 +1502,7 @@ class TestUpdatePublicModelGroups: mock_proxy_config.get_config = mock_get_config mock_proxy_config.save_config = AsyncMock() - admin_user = UserAPIKeyAuth( - user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + admin_user = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) request = UpdatePublicModelGroupsRequest(model_groups=new_models) @@ -1561,9 +1558,7 @@ class TestUpdatePublicModelGroups: mock_proxy_config.get_config = mock_get_config mock_proxy_config.save_config = AsyncMock() - admin_user = UserAPIKeyAuth( - user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + admin_user = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) request = UpdateUsefulLinksRequest(useful_links=new_links) @@ -1728,9 +1723,7 @@ class TestTeamModelSiblingRouting: ) # Global deployment should be accessible when team_id is provided - deployments = router._get_all_deployments( - model_name="global-gpt-4o", team_id="teamA" - ) + deployments = router._get_all_deployments(model_name="global-gpt-4o", team_id="teamA") assert len(deployments) == 1 assert deployments[0]["model_name"] == "global-gpt-4o" @@ -1779,9 +1772,7 @@ class TestTeamModelUpdate: patch( "litellm.proxy.management_endpoints.model_management_endpoints.team_model_add" ) as mock_team_model_add, - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.update_team" - ) as mock_update_team, + patch("litellm.proxy.management_endpoints.model_management_endpoints.update_team") as mock_update_team, ): result = await _update_team_model_in_db( db_model=db_model, @@ -1812,9 +1803,7 @@ class TestTeamModelUpdate: db_model = Deployment( model_name="model_name_team_123_uuid1", litellm_params=LiteLLM_Params(model="azure/gpt-4o-mini"), - model_info=ModelInfo( - team_id="team_123", team_public_model_name="old-public-name" - ), + model_info=ModelInfo(team_id="team_123", team_public_model_name="old-public-name"), ) # Create a sibling deployment that still uses the old public name @@ -1825,9 +1814,7 @@ class TestTeamModelUpdate: "team_public_model_name": "old-public-name", } - prisma_client = MockPrismaClient( - team_exists=True, sibling_deployments=[sibling_deployment] - ) + prisma_client = MockPrismaClient(team_exists=True, sibling_deployments=[sibling_deployment]) patch_data = updateDeployment( model_name="new-public-name", @@ -1840,12 +1827,8 @@ class TestTeamModelUpdate: ) with ( - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete" - ) as mock_delete, - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.team_model_add" - ) as mock_add, + patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete") as mock_delete, + patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_add") as mock_add, ): await _update_existing_team_model_assignment( team_id="team_123", @@ -1885,12 +1868,8 @@ class TestTeamModelUpdate: ) with ( - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete" - ) as mock_delete, - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.team_model_add" - ) as mock_add, + patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete") as mock_delete, + patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_add") as mock_add, ): await _update_existing_team_model_assignment( team_id="team_123", @@ -1944,7 +1923,9 @@ class TestTeamModelUpdate: "litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.allow_team_model_action", AsyncMock(return_value=True), ), - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: team models are premium-gated through a proxy global with no injection seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: team models are premium-gated through a proxy global with no injection seam patch( # test-quality-ok: the team list write is the collaborator whose ordering is asserted "litellm.proxy.management_endpoints.model_management_endpoints.team_model_add", side_effect=team_add, @@ -1984,20 +1965,14 @@ class TestTeamModelUpdate: db_model = Deployment( model_name="model_name_team_123_uuid1", litellm_params=LiteLLM_Params(model="azure/gpt-4o-mini"), - model_info=ModelInfo( - team_id="team_123", team_public_model_name="old-public-name" - ), + model_info=ModelInfo(team_id="team_123", team_public_model_name="old-public-name"), ) sibling_deployment = MagicMock() sibling_deployment.model_name = "model_name_team_123_uuid2" - sibling_deployment.model_info = ( - '{"team_id":"team_123","team_public_model_name":"old-public-name"}' - ) + sibling_deployment.model_info = '{"team_id":"team_123","team_public_model_name":"old-public-name"}' - prisma_client = MockPrismaClient( - team_exists=True, sibling_deployments=[sibling_deployment] - ) + prisma_client = MockPrismaClient(team_exists=True, sibling_deployments=[sibling_deployment]) patch_data = updateDeployment( model_name="new-public-name", @@ -2010,12 +1985,8 @@ class TestTeamModelUpdate: ) with ( - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete" - ) as mock_delete, - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.team_model_add" - ) as mock_add, + patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete") as mock_delete, + patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_add") as mock_add, ): await _update_existing_team_model_assignment( team_id="team_123", @@ -2092,10 +2063,7 @@ class TestTeamModelUpdate: ), ) - assert ( - _get_public_model_name(patch_data=patch_data, db_model=db_model) - == "gpt-5.2-low-rpm-testing" - ) + assert _get_public_model_name(patch_data=patch_data, db_model=db_model) == "gpt-5.2-low-rpm-testing" def test_get_public_model_name_preserves_db_public_name_when_internal_name_unchanged( self, @@ -2122,10 +2090,7 @@ class TestTeamModelUpdate: model_info=ModelInfo(team_id="test-team"), ) - assert ( - _get_public_model_name(patch_data=patch_data, db_model=db_model) - == "gpt-5.2-low-rpm-testing" - ) + assert _get_public_model_name(patch_data=patch_data, db_model=db_model) == "gpt-5.2-low-rpm-testing" def test_get_public_model_name_allows_top_level_rename(self): """A genuine rename via the top-level model_name field (no @@ -2150,10 +2115,7 @@ class TestTeamModelUpdate: model_info=ModelInfo(team_id="test-team"), ) - assert ( - _get_public_model_name(patch_data=patch_data, db_model=db_model) - == "new-public-name" - ) + assert _get_public_model_name(patch_data=patch_data, db_model=db_model) == "new-public-name" def test_get_public_model_name_top_level_rename_wins_over_stale_model_info(self): """Regression (codex review): on a dashboard rename the UI sends the new @@ -2170,9 +2132,7 @@ class TestTeamModelUpdate: db_model = Deployment( model_name="model_name_team-a_abc123", litellm_params=LiteLLM_Params(model="azure/gpt-4.1"), - model_info=ModelInfo( - team_id="team-a", team_public_model_name="old-public-name" - ), + model_info=ModelInfo(team_id="team-a", team_public_model_name="old-public-name"), ) patch_data = updateDeployment( model_name="new-public-name", @@ -2182,10 +2142,7 @@ class TestTeamModelUpdate: ), ) - assert ( - _get_public_model_name(patch_data=patch_data, db_model=db_model) - == "new-public-name" - ) + assert _get_public_model_name(patch_data=patch_data, db_model=db_model) == "new-public-name" def test_get_public_model_name_falls_back_to_db_public_name(self): """When patch_data carries no name hints at all (neither model_name @@ -2208,10 +2165,7 @@ class TestTeamModelUpdate: model_info=ModelInfo(team_id="test-team"), ) - assert ( - _get_public_model_name(patch_data=patch_data, db_model=db_model) - == "gpt-5.2-low-rpm-testing" - ) + assert _get_public_model_name(patch_data=patch_data, db_model=db_model) == "gpt-5.2-low-rpm-testing" def test_get_public_model_name_last_resort_returns_db_model_name(self): """Legacy rows may have no team_public_model_name anywhere; the @@ -2231,10 +2185,7 @@ class TestTeamModelUpdate: model_info=ModelInfo(team_id="test-team"), ) - assert ( - _get_public_model_name(patch_data=patch_data, db_model=db_model) - == "legacy-model" - ) + assert _get_public_model_name(patch_data=patch_data, db_model=db_model) == "legacy-model" def test_get_public_model_name_ignores_different_internal_shape_name(self): """A stale client may PATCH an internal-shaped model_name that does not @@ -2258,10 +2209,7 @@ class TestTeamModelUpdate: model_info=ModelInfo(team_id="test-team"), ) - assert ( - _get_public_model_name(patch_data=patch_data, db_model=db_model) - == "gpt-5.2-low-rpm-testing" - ) + assert _get_public_model_name(patch_data=patch_data, db_model=db_model) == "gpt-5.2-low-rpm-testing" def test_get_public_model_name_ignores_internal_shape_patch_public(self): """If a corrupted row round-trips an internal-shaped value in @@ -2287,10 +2235,7 @@ class TestTeamModelUpdate: ), ) - assert ( - _get_public_model_name(patch_data=patch_data, db_model=db_model) - == "gpt-5.2-low-rpm-testing" - ) + assert _get_public_model_name(patch_data=patch_data, db_model=db_model) == "gpt-5.2-low-rpm-testing" @pytest.mark.asyncio async def test_dashboard_edit_preserves_public_name_and_acl(self): @@ -2358,9 +2303,7 @@ class TestTeamModelUpdate: # the merged model_info written to the DB must keep the public name model_info_json = result.get("model_info", "") parsed_model_info = json.loads(model_info_json) - assert ( - parsed_model_info.get("team_public_model_name") == "gpt-5.2-low-rpm-testing" - ) + assert parsed_model_info.get("team_public_model_name") == "gpt-5.2-low-rpm-testing" # the internal model_name must not have been overwritten (caller # intentionally clears patch_data.model_name so the DB row's name @@ -2402,9 +2345,7 @@ class TestModelInfoEndpoint: model_info=ModelInfo(id="gpt-4"), ) - result = await model_info( - model_id="gpt-4", user_api_key_dict=user_api_key_dict - ) + result = await model_info(model_id="gpt-4", user_api_key_dict=user_api_key_dict) assert result["id"] == "gpt-4" assert result["object"] == "model" @@ -2479,9 +2420,7 @@ class TestModelInfoEndpoint: model_info=ModelInfo(id="team-model-1"), ) - result = await model_info( - model_id="team-model-1", user_api_key_dict=user_api_key_dict - ) + result = await model_info(model_id="team-model-1", user_api_key_dict=user_api_key_dict) assert result["id"] == "team-model-1" assert result["object"] == "model" @@ -2513,9 +2452,7 @@ class TestAddAndDeleteModelLifecycle: ) model_id = "lifecycle-test-model-123" - admin_user = UserAPIKeyAuth( - user_id="test-admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + admin_user = UserAPIKeyAuth(user_id="test-admin", user_role=LitellmUserRoles.PROXY_ADMIN) # Build a real LiteLLM_ProxyModelTable for the DB mock to return db_row = LiteLLM_ProxyModelTable( @@ -2532,9 +2469,7 @@ class TestAddAndDeleteModelLifecycle: mock_prisma.db.litellm_proxymodeltable = AsyncMock() mock_prisma.db.query_raw = AsyncMock(return_value=[]) mock_prisma.db.litellm_proxymodeltable.create = AsyncMock(return_value=db_row) - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=db_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.delete = AsyncMock(return_value=db_row) mock_proxy_config = MagicMock() @@ -2558,14 +2493,11 @@ class TestAddAndDeleteModelLifecycle: patch(f"{_PS}.llm_router", mock_router), patch(_ENCRYPT, side_effect=lambda value, **kwargs: value), ): - # --- ADD --- add_result = await add_new_model( model_params=Deployment( model_name="lifecycle-model", - litellm_params=LiteLLM_Params( - model="openai/gpt-4.1-nano", api_key="fake-key" - ), + litellm_params=LiteLLM_Params(model="openai/gpt-4.1-nano", api_key="fake-key"), model_info={"id": model_id}, ), user_api_key_dict=admin_user, @@ -2580,9 +2512,7 @@ class TestAddAndDeleteModelLifecycle: assert "deleted successfully" in delete_result["message"] # --- DELETE again should fail (model not found) --- - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=None - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=None) from litellm.proxy.proxy_server import ProxyException with pytest.raises(ProxyException) as exc_info: @@ -2644,24 +2574,18 @@ class TestDeleteTeamBYOKModelGhost: mock_prisma.db = MagicMock() mock_prisma.db.litellm_proxymodeltable = AsyncMock() mock_prisma.db.query_raw = AsyncMock(return_value=[]) - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=db_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.delete = AsyncMock(return_value=db_row) # After the row delete no team deployment remains -> nothing backs the public name. mock_prisma.db.litellm_proxymodeltable.find_many = AsyncMock(return_value=[]) mock_prisma.db.litellm_teamtable = AsyncMock() mock_prisma.db.litellm_teamtable.find_unique = AsyncMock(return_value=team_row) - mock_prisma.db.litellm_teamtable.update = AsyncMock( - return_value=updated_team_row - ) + mock_prisma.db.litellm_teamtable.update = AsyncMock(return_value=updated_team_row) # Team BYOK models have no alias row; delete_team_model_alias finds nothing. mock_prisma.db.litellm_modeltable = AsyncMock() mock_prisma.db.litellm_modeltable.find_many = AsyncMock(return_value=[]) - admin_user = UserAPIKeyAuth( - user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + admin_user = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) _PS = "litellm.proxy.proxy_server" _MOD = "litellm.proxy.management_endpoints.model_management_endpoints" @@ -2727,9 +2651,7 @@ class TestDeleteTeamBYOKModelGhost: mock_prisma.db = MagicMock() mock_prisma.db.litellm_proxymodeltable = AsyncMock() mock_prisma.db.query_raw = AsyncMock(return_value=[]) - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=db_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.delete = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.find_many = AsyncMock(return_value=[]) mock_prisma.db.litellm_teamtable = AsyncMock() @@ -2739,9 +2661,7 @@ class TestDeleteTeamBYOKModelGhost: # No alias row matches -> delete_team_model_alias returns nothing, but it still ran. mock_prisma.db.litellm_modeltable.find_many = AsyncMock(return_value=[]) - admin_user = UserAPIKeyAuth( - user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + admin_user = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) _PS = "litellm.proxy.proxy_server" _MOD = "litellm.proxy.management_endpoints.model_management_endpoints" @@ -2804,25 +2724,17 @@ class TestDeleteTeamBYOKModelGhost: mock_prisma.db = MagicMock() mock_prisma.db.litellm_proxymodeltable = AsyncMock() mock_prisma.db.query_raw = AsyncMock(return_value=[]) - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=deleted_row - ) - mock_prisma.db.litellm_proxymodeltable.delete = AsyncMock( - return_value=deleted_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=deleted_row) + mock_prisma.db.litellm_proxymodeltable.delete = AsyncMock(return_value=deleted_row) # After the deleted replica's row is gone, the sibling still backs the public name. - mock_prisma.db.litellm_proxymodeltable.find_many = AsyncMock( - return_value=[sibling_row] - ) + mock_prisma.db.litellm_proxymodeltable.find_many = AsyncMock(return_value=[sibling_row]) mock_prisma.db.litellm_teamtable = AsyncMock() mock_prisma.db.litellm_teamtable.find_unique = AsyncMock(return_value=team_row) mock_prisma.db.litellm_teamtable.update = AsyncMock(return_value=team_row) mock_prisma.db.litellm_modeltable = AsyncMock() mock_prisma.db.litellm_modeltable.find_many = AsyncMock(return_value=[]) - admin_user = UserAPIKeyAuth( - user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + admin_user = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) _PS = "litellm.proxy.proxy_server" _MOD = "litellm.proxy.management_endpoints.model_management_endpoints" @@ -2880,9 +2792,7 @@ class TestDeleteTeamBYOKModelGhost: members_with_roles=[Member(user_id="admin", role="admin")], models=[public_name], ) - alias_row = MagicMock( - id="alias-row-1", model_aliases={public_name: internal_name} - ) + alias_row = MagicMock(id="alias-row-1", model_aliases={public_name: internal_name}) alias_row.team = MagicMock() alias_row.team.team_id = team_id @@ -2890,26 +2800,20 @@ class TestDeleteTeamBYOKModelGhost: mock_prisma.db = MagicMock() mock_prisma.db.litellm_proxymodeltable = AsyncMock() mock_prisma.db.query_raw = AsyncMock(return_value=[]) - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=db_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.delete = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.find_many = AsyncMock(return_value=[]) mock_prisma.db.litellm_teamtable = AsyncMock() mock_prisma.db.litellm_teamtable.find_unique = AsyncMock(return_value=team_row) mock_prisma.db.litellm_teamtable.update = AsyncMock(return_value=team_row) mock_prisma.db.litellm_modeltable = AsyncMock() - mock_prisma.db.litellm_modeltable.find_many = AsyncMock( - return_value=[alias_row] - ) + mock_prisma.db.litellm_modeltable.find_many = AsyncMock(return_value=[alias_row]) mock_prisma.db.litellm_modeltable.update = AsyncMock() mock_router = MagicMock() mock_router.model_name_to_deployment_indices = {public_name: [0]} - admin_user = UserAPIKeyAuth( - user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + admin_user = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) _PS = "litellm.proxy.proxy_server" _MOD = "litellm.proxy.management_endpoints.model_management_endpoints" @@ -2972,9 +2876,7 @@ class TestDeleteTeamBYOKModelGhost: mock_prisma.db = MagicMock() mock_prisma.db.litellm_proxymodeltable = AsyncMock() mock_prisma.db.query_raw = AsyncMock(return_value=[]) - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=db_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.delete = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.find_many = AsyncMock(return_value=[]) mock_prisma.db.litellm_teamtable = AsyncMock() @@ -2987,9 +2889,7 @@ class TestDeleteTeamBYOKModelGhost: mock_router = MagicMock() mock_router.model_name_to_deployment_indices = {internal_name: [0]} - admin_user = UserAPIKeyAuth( - user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + admin_user = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) _PS = "litellm.proxy.proxy_server" _MOD = "litellm.proxy.management_endpoints.model_management_endpoints" @@ -3042,9 +2942,7 @@ class TestDeleteModelTeamAuth: mock_prisma.db = MagicMock() mock_prisma.db.litellm_proxymodeltable = AsyncMock() mock_prisma.db.query_raw = AsyncMock(return_value=[]) - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=db_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.delete = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.find_many = AsyncMock(return_value=[]) # The team is gone -> every team lookup returns None. @@ -3066,9 +2964,7 @@ class TestDeleteModelTeamAuth: model_id = "orphaned-byok-1" mock_prisma = self._orphaned_model_mocks(team_id, model_id) - admin_user = UserAPIKeyAuth( - user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN - ) + admin_user = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) _PS = "litellm.proxy.proxy_server" _MOD = "litellm.proxy.management_endpoints.model_management_endpoints" @@ -3104,9 +3000,7 @@ class TestDeleteModelTeamAuth: model_id = "orphaned-byok-2" mock_prisma = self._orphaned_model_mocks(team_id, model_id) - non_admin = UserAPIKeyAuth( - user_id="someone", user_role=LitellmUserRoles.INTERNAL_USER - ) + non_admin = UserAPIKeyAuth(user_id="someone", user_role=LitellmUserRoles.INTERNAL_USER) _PS = "litellm.proxy.proxy_server" _MOD = "litellm.proxy.management_endpoints.model_management_endpoints" @@ -3161,9 +3055,7 @@ class TestDeleteModelTeamAuth: mock_prisma.db = MagicMock() mock_prisma.db.litellm_proxymodeltable = AsyncMock() mock_prisma.db.query_raw = AsyncMock(return_value=[]) - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=db_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.delete = AsyncMock(return_value=db_row) mock_prisma.db.litellm_proxymodeltable.find_many = AsyncMock(return_value=[]) mock_prisma.db.litellm_teamtable = AsyncMock() @@ -3173,9 +3065,7 @@ class TestDeleteModelTeamAuth: # A team member who is not the team admin: rejected before the delete runs, # so the only team lookup is the single one inside the auth check. - non_admin = UserAPIKeyAuth( - user_id="someone", user_role=LitellmUserRoles.INTERNAL_USER - ) + non_admin = UserAPIKeyAuth(user_id="someone", user_role=LitellmUserRoles.INTERNAL_USER) _PS = "litellm.proxy.proxy_server" _MOD = "litellm.proxy.management_endpoints.model_management_endpoints" @@ -3369,15 +3259,11 @@ class TestDeleteTeamModels: prisma = _TxPrismaClient(rows) router = _RecordingRouter(prisma.events) - await delete_team_models( - team_ids=["team_a", "team_b"], prisma_client=prisma, llm_router=router - ) + await delete_team_models(team_ids=["team_a", "team_b"], prisma_client=prisma, llm_router=router) commit_idx = prisma.events.index(("commit",)) router_indices = [i for i, e in enumerate(prisma.events) if e[0] == "router"] - delete_indices = [ - i for i, e in enumerate(prisma.events) if e[0] == "delete_many" - ] + delete_indices = [i for i, e in enumerate(prisma.events) if e[0] == "delete_many"] assert router_indices, "router was never synced" assert all(i > commit_idx for i in router_indices) assert all(i < commit_idx for i in delete_indices) @@ -3393,9 +3279,7 @@ class TestDeleteTeamModels: prisma = _TxPrismaClient([mine, intruder]) router = _RecordingRouter(prisma.events) - deleted = await delete_team_models( - team_ids=["team_a"], prisma_client=prisma, llm_router=router - ) + deleted = await delete_team_models(team_ids=["team_a"], prisma_client=prisma, llm_router=router) assert deleted == ["a1"] assert router.deleted == ["a1"] @@ -3405,9 +3289,7 @@ class TestDeleteTeamModels: prisma = _TxPrismaClient([]) router = _RecordingRouter(prisma.events) - deleted = await delete_team_models( - team_ids=["team_a"], prisma_client=prisma, llm_router=router - ) + deleted = await delete_team_models(team_ids=["team_a"], prisma_client=prisma, llm_router=router) assert deleted == [] assert router.deleted == [] @@ -3418,9 +3300,7 @@ class TestDeleteTeamModels: rows = [_model_row("a1", "team_a")] prisma = _TxPrismaClient(rows) - deleted = await delete_team_models( - team_ids=["team_a"], prisma_client=prisma, llm_router=None - ) + deleted = await delete_team_models(team_ids=["team_a"], prisma_client=prisma, llm_router=None) assert deleted == ["a1"] assert any(e[0] == "delete_many" for e in prisma.events) @@ -3626,9 +3506,7 @@ class TestUpdateDBModelClearPricing: result = update_db_model( db_model=_build_db_model_with_pricing(), - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams(input_cost_per_token=None) - ), + updated_patch=updateDeployment(litellm_params=updateLiteLLMParams(input_cost_per_token=None)), ) params = json.loads(result["litellm_params"]) @@ -3647,9 +3525,7 @@ class TestUpdateDBModelClearPricing: result = update_db_model( db_model=_build_db_model_with_pricing(), - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams(output_cost_per_token=None) - ), + updated_patch=updateDeployment(litellm_params=updateLiteLLMParams(output_cost_per_token=None)), ) params = json.loads(result["litellm_params"]) @@ -3665,9 +3541,7 @@ class TestUpdateDBModelClearPricing: result = update_db_model( db_model=_build_db_model_with_pricing(), - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams(input_cost_per_token=0.000005) - ), + updated_patch=updateDeployment(litellm_params=updateLiteLLMParams(input_cost_per_token=0.000005)), ) params = json.loads(result["litellm_params"]) @@ -3682,9 +3556,7 @@ class TestUpdateDBModelClearPricing: result = update_db_model( db_model=_build_db_model_with_pricing(), - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams(output_cost_per_token=0.000007) - ), + updated_patch=updateDeployment(litellm_params=updateLiteLLMParams(output_cost_per_token=0.000007)), ) params = json.loads(result["litellm_params"]) @@ -3719,9 +3591,7 @@ class TestUpdateDBModelClearPricing: # or any other non-pricing field from the merged dict. result = update_db_model( db_model=db_model, - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams(api_base=None) - ), + updated_patch=updateDeployment(litellm_params=updateLiteLLMParams(api_base=None)), ) info = json.loads(result["model_info"]) @@ -3756,9 +3626,7 @@ class TestUpdateDBModelClearPricing: params = json.loads(result["litellm_params"]) info = json.loads(result["model_info"]) assert "input_cost_per_token" not in params - assert ( - "input_cost_per_token" not in info - ), "model_info passthrough must not resurrect the cleared override" + assert "input_cost_per_token" not in info, "model_info passthrough must not resurrect the cleared override" def test_clear_via_model_info_clears_both_blobs(self): """The mirror works in the reverse direction too: nulling a pricing field @@ -3770,9 +3638,7 @@ class TestUpdateDBModelClearPricing: result = update_db_model( db_model=_build_db_model_with_pricing(), - updated_patch=updateDeployment( - model_info=ModelInfo(id="dep-pricing-0", input_cost_per_token=None) - ), + updated_patch=updateDeployment(model_info=ModelInfo(id="dep-pricing-0", input_cost_per_token=None)), ) params = json.loads(result["litellm_params"]) @@ -3804,9 +3670,7 @@ class TestUpdateDBModelClearPricing: result = update_db_model( db_model=db_model, - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams(cache_read_input_token_cost=None) - ), + updated_patch=updateDeployment(litellm_params=updateLiteLLMParams(cache_read_input_token_cost=None)), ) params = json.loads(result["litellm_params"]) @@ -3838,9 +3702,7 @@ class TestUpdateDBModelClearPricing: result = update_db_model( db_model=db_model, - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams(cache_creation_input_token_cost=None) - ), + updated_patch=updateDeployment(litellm_params=updateLiteLLMParams(cache_creation_input_token_cost=None)), ) params = json.loads(result["litellm_params"]) @@ -3874,9 +3736,7 @@ class TestUpdateDBModelClearPricing: result = update_db_model( db_model=db_model, - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams(cache_read_input_token_cost=None) - ), + updated_patch=updateDeployment(litellm_params=updateLiteLLMParams(cache_read_input_token_cost=None)), ) params = json.loads(result["litellm_params"]) @@ -4383,9 +4243,7 @@ class TestModelInfoCostMapEchoFilter: for deployment_entry in (litellm.get_model_info(deployment_model),) for key_entry in (litellm.get_model_info(key_model),) for differing_fields in ( - frozenset( - k for k in deployment_entry if k in key_entry and deployment_entry[k] != key_entry[k] - ), + frozenset(k for k in deployment_entry if k in key_entry and deployment_entry[k] != key_entry[k]), ) if differing_fields ) @@ -4478,9 +4336,7 @@ class TestUpdateDBModelClearCacheControlInjectionPoints: ), model_info=ModelInfo(id="dep-cache-0"), ) - patch = updateDeployment( - litellm_params=updateLiteLLMParams(cache_control_injection_points=None) - ) + patch = updateDeployment(litellm_params=updateLiteLLMParams(cache_control_injection_points=None)) result = update_db_model(db_model=db_model, updated_patch=patch) @@ -4526,11 +4382,12 @@ class TestUpdateDBModelClearCredentialName: ), model_info=ModelInfo(id="dep-cred-1", team_id="team-keep", access_groups=["prod"]), ) - update_patch: Final = updateDeployment( - litellm_params=updateLiteLLMParams(litellm_credential_name=None) - ) + update_patch: Final = updateDeployment(litellm_params=updateLiteLLMParams(litellm_credential_name=None)) - with patch("litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", side_effect=lambda value, **kwargs: value): + with patch( + "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", + side_effect=lambda value, **kwargs: value, + ): result: Final = update_db_model(db_model=db_model, updated_patch=update_patch) params: Final = json.loads(result["litellm_params"]) @@ -4559,7 +4416,10 @@ class TestUpdateDBModelClearCredentialName: ) update_patch: Final = updateDeployment(litellm_params=updateLiteLLMParams(tpm=10)) - with patch("litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", side_effect=lambda value, **kwargs: value): + with patch( + "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", + side_effect=lambda value, **kwargs: value, + ): result: Final = update_db_model(db_model=db_model, updated_patch=update_patch) params: Final = json.loads(result["litellm_params"]) @@ -4574,11 +4434,12 @@ class TestUpdateDBModelClearCredentialName: litellm_params=LiteLLM_Params(model="openai/gpt-4o", api_base="https://api.openai.com/v1"), model_info=ModelInfo(id="dep-cred-1", team_id="team-keep", access_groups=["prod"]), ) - update_patch: Final = updateDeployment( - litellm_params=updateLiteLLMParams(litellm_credential_name=None) - ) + update_patch: Final = updateDeployment(litellm_params=updateLiteLLMParams(litellm_credential_name=None)) - with patch("litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", side_effect=lambda value, **kwargs: value): + with patch( + "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", + side_effect=lambda value, **kwargs: value, + ): result: Final = update_db_model(db_model=db_model, updated_patch=update_patch) params: Final = json.loads(result["litellm_params"]) @@ -4606,7 +4467,10 @@ class TestUpdateDBModelClearCredentialName: ) ) - with patch("litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", side_effect=lambda value, **kwargs: value): + with patch( + "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", + side_effect=lambda value, **kwargs: value, + ): result: Final = update_db_model(db_model=db_model, updated_patch=update_patch) params: Final = json.loads(result["litellm_params"]) @@ -4633,7 +4497,10 @@ class TestUpdateDBModelClearCredentialName: litellm_params=updateLiteLLMParams(litellm_credential_name="other-credential") ) - with patch("litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", side_effect=lambda value, **kwargs: value): + with patch( + "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", + side_effect=lambda value, **kwargs: value, + ): result: Final = update_db_model(db_model=db_model, updated_patch=update_patch) params: Final = json.loads(result["litellm_params"]) @@ -4712,9 +4579,7 @@ class TestPatchModelCredentialName: "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", side_effect=lambda value, **kwargs: value, ), - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.raise_if_reload_degraded_serving" - ), + patch("litellm.proxy.management_endpoints.model_management_endpoints.raise_if_reload_degraded_serving"), patch( "litellm.proxy.management_endpoints.model_management_endpoints.create_object_audit_log", new=AsyncMock(), @@ -4980,9 +4845,7 @@ class TestPatchModelBlockedAuthGate: existing_row.model_dump_json.return_value = "{}" mock_prisma = MagicMock() - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=existing_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=existing_row) with ( patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), @@ -5023,12 +4886,8 @@ class TestPatchModelBlockedAuthGate: updated_row.model_dump_json.return_value = "{}" mock_prisma = MagicMock() - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=existing_row - ) - mock_prisma.db.litellm_proxymodeltable.update = AsyncMock( - return_value=updated_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=existing_row) + mock_prisma.db.litellm_proxymodeltable.update = AsyncMock(return_value=updated_row) with ( patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), @@ -5041,9 +4900,7 @@ class TestPatchModelBlockedAuthGate: ), patch( # test-quality-ok: [TQ008] isolate persistence from router reload implementation "litellm.proxy.management_endpoints.model_management_endpoints.clear_cache", - new=AsyncMock( - return_value=ReconcileOutcome(still_desired=None, live_after=None) - ), + new=AsyncMock(return_value=ReconcileOutcome(still_desired=None, live_after=None)), ), ): result = await patch_model( @@ -5078,25 +4935,29 @@ class TestPatchModelRowDeletedBeforeWrite: existing_row.model_dump_json.return_value = "{}" mock_prisma = MagicMock() - mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock( - return_value=existing_row - ) + mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=existing_row) mock_prisma.db.litellm_proxymodeltable.update = AsyncMock(return_value=None) with ( - patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), # test-quality-ok: proxy_server module global is the endpoint's only injection point - patch("litellm.proxy.proxy_server.llm_router", MagicMock(**{"get_model_ids.return_value": ["m1"]})), # test-quality-ok: proxy_server module global is the endpoint's only injection point - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: proxy_server module global is the endpoint's only injection point - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: proxy_server module global is the endpoint's only injection point + patch( + "litellm.proxy.proxy_server.prisma_client", mock_prisma + ), # test-quality-ok: proxy_server module global is the endpoint's only injection point + patch( + "litellm.proxy.proxy_server.llm_router", MagicMock(**{"get_model_ids.return_value": ["m1"]}) + ), # test-quality-ok: proxy_server module global is the endpoint's only injection point + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: proxy_server module global is the endpoint's only injection point + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: proxy_server module global is the endpoint's only injection point patch( # test-quality-ok: stubs the auth gate so the test exercises the not-found branch under test "litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call", new=AsyncMock(return_value=None), ), patch( # test-quality-ok: stubs the cache write so the test observes only the DB result handling "litellm.proxy.management_endpoints.model_management_endpoints.clear_cache", - new=AsyncMock( - return_value=ReconcileOutcome(still_desired=None, live_after=None) - ), + new=AsyncMock(return_value=ReconcileOutcome(still_desired=None, live_after=None)), ), ): with pytest.raises(ProxyException) as exc_info: @@ -5180,9 +5041,7 @@ class TestWriteSurfacesReloadDrop: ) with pytest.raises(ProxyException, match="m-gone"): - raise_if_reload_degraded_serving( - before=frozenset(), written_models=[("m-gone", None)], action="update" - ) + raise_if_reload_degraded_serving(before=frozenset(), written_models=[("m-gone", None)], action="update") with pytest.raises(ProxyException, match="m-collateral"): raise_if_reload_degraded_serving( @@ -5297,10 +5156,7 @@ class TestConcurrentModelWritesDoNotEvictEachOther: config = ProxyConfig() await asyncio.gather( - *[ - config.add_deployment(prisma_client=MagicMock(), proxy_logging_obj=MagicMock()) - for _ in range(5) - ] + *[config.add_deployment(prisma_client=MagicMock(), proxy_logging_obj=MagicMock()) for _ in range(5)] ) assert observed_max == 1 @@ -5529,9 +5385,7 @@ class TestDeleteEvictionsHoldTheReconcileLock: ) async def call() -> None: - await delete_team_models( - team_ids=["team-1"], prisma_client=prisma, llm_router=router - ) + await delete_team_models(team_ids=["team-1"], prisma_client=prisma, llm_router=router) await self._assert_evicts_under_lock(monkeypatch, call, model_id, config) router.delete_deployment.assert_called_once_with(id=model_id) @@ -6100,7 +5954,9 @@ class TestStrategyRouterWriteValidation: **_FORECAST_BASE, "classifier_type": "capability", "capability_classifier_config": { - "efficient_tier": "SIMPLE", "capable_tier": "REASONING", "base_threshold": 0.7, + "efficient_tier": "SIMPLE", + "capable_tier": "REASONING", + "base_threshold": 0.7, }, } _FUSE = { @@ -6108,8 +5964,10 @@ class TestStrategyRouterWriteValidation: "classifier_type": "llm_v2", "adaptive": False, "llm_v2_config": { - "efficient_profile": "Small solver", "capable_profile": "Large solver", - "harness": "One attempt", "max_quality_gap": 0.05, + "efficient_profile": "Small solver", + "capable_profile": "Large solver", + "harness": "One attempt", + "max_quality_gap": 0.05, }, } _CUSTOM_TIERS = { @@ -6160,17 +6018,17 @@ class TestStrategyRouterWriteValidation: ("no-config", _V2, _V2), ], ) - def test_effective_complexity_router_config( - self, incoming: object, existing: object, expected: object - ) -> None: + def test_effective_complexity_router_config(self, incoming: object, existing: object, expected: object) -> None: """A write is judged on the config it leaves on the row: the incoming one when it carries one, else the stored one.""" from litellm.proxy.management_endpoints.model_management_endpoints import ( _effective_complexity_router_config, ) from litellm.types.router import updateLiteLLMParams - incoming_params = None if incoming is None else updateLiteLLMParams( - complexity_router_config=None if incoming == "no-config" else incoming + incoming_params = ( + None + if incoming is None + else updateLiteLLMParams(complexity_router_config=None if incoming == "no-config" else incoming) ) existing_params = None if existing is None else updateLiteLLMParams(complexity_router_config=existing) assert _effective_complexity_router_config(incoming_params, existing_params) == expected @@ -6179,32 +6037,207 @@ class TestStrategyRouterWriteValidation: @pytest.mark.parametrize( "limit,effective_params,db_models,config_config,model_id,expected", [ - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _CAPABILITY}, ["auto_router/complexity_router"], None, None, "refused"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _CAPABILITY}, [], _CAPABILITY, None, "refused"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _CAPABILITY}, [], _FUSE, None, "reserved"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _CAPABILITY}, [], None, "held-id", "reserved"), - (None, {"model": "auto_router/complexity_router", "complexity_router_config": _CAPABILITY}, ["auto_router/complexity_router"], _CAPABILITY, None, "plain"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _FUSE}, ["auto_router/complexity_router"], None, None, "refused"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _FUSE}, [], _FUSE, None, "refused"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _FUSE}, [], _CAPABILITY, None, "reserved"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _FUSE}, [], None, "held-id", "reserved"), - (None, {"model": "auto_router/complexity_router", "complexity_router_config": _FUSE}, ["auto_router/complexity_router"], _FUSE, None, "plain"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, ["auto_router/complexity_router"], None, None, "refused"), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _CAPABILITY}, + ["auto_router/complexity_router"], + None, + None, + "refused", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _CAPABILITY}, + [], + _CAPABILITY, + None, + "refused", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _CAPABILITY}, + [], + _FUSE, + None, + "reserved", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _CAPABILITY}, + [], + None, + "held-id", + "reserved", + ), + ( + None, + {"model": "auto_router/complexity_router", "complexity_router_config": _CAPABILITY}, + ["auto_router/complexity_router"], + _CAPABILITY, + None, + "plain", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _FUSE}, + ["auto_router/complexity_router"], + None, + None, + "refused", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _FUSE}, + [], + _FUSE, + None, + "refused", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _FUSE}, + [], + _CAPABILITY, + None, + "reserved", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _FUSE}, + [], + None, + "held-id", + "reserved", + ), + ( + None, + {"model": "auto_router/complexity_router", "complexity_router_config": _FUSE}, + ["auto_router/complexity_router"], + _FUSE, + None, + "plain", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, + ["auto_router/complexity_router"], + None, + None, + "refused", + ), (1, {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, [], _V2, None, "refused"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, [], None, None, "reserved"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, [], None, "held-id", "reserved"), - (2, {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, ["auto_router/complexity_router"], None, None, "reserved"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _CUSTOM_TIERS}, ["openai/gpt-4o"], None, None, "reserved"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _CUSTOM_TIERS}, [], _CUSTOM_PROMPT, None, "refused"), - (1, {"model": "openai/gpt-4o", "complexity_router_config": _CUSTOM_TIERS}, ["auto_router/complexity_router"], None, None, "plain"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _V1}, ["auto_router/complexity_router"], _V2, None, "plain"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": None}, ["auto_router/complexity_router"], _V2, None, "plain"), - (None, {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, ["auto_router/complexity_router"], _V2, None, "plain"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _TIER_LABELS_ONLY}, ["auto_router/complexity_router"], _CUSTOM_TIERS, None, "plain"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _CUSTOM_PROMPT}, ["auto_router/complexity_router"], None, None, "refused"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _OPERATOR_EXAMPLES}, [], _CUSTOM_TIERS, None, "refused"), - (1, {"model": "auto_router/complexity_router", "complexity_router_config": _OPERATOR_OPENING_PROMPT}, [], _CUSTOM_PROMPT, None, "refused"), - (None, {"model": "auto_router/complexity_router", "complexity_router_config": _OPERATOR_EXAMPLES}, ["auto_router/complexity_router"], _CUSTOM_TIERS, None, "plain"), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, + [], + None, + None, + "reserved", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, + [], + None, + "held-id", + "reserved", + ), + ( + 2, + {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, + ["auto_router/complexity_router"], + None, + None, + "reserved", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _CUSTOM_TIERS}, + ["openai/gpt-4o"], + None, + None, + "reserved", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _CUSTOM_TIERS}, + [], + _CUSTOM_PROMPT, + None, + "refused", + ), + ( + 1, + {"model": "openai/gpt-4o", "complexity_router_config": _CUSTOM_TIERS}, + ["auto_router/complexity_router"], + None, + None, + "plain", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _V1}, + ["auto_router/complexity_router"], + _V2, + None, + "plain", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": None}, + ["auto_router/complexity_router"], + _V2, + None, + "plain", + ), + ( + None, + {"model": "auto_router/complexity_router", "complexity_router_config": _V2}, + ["auto_router/complexity_router"], + _V2, + None, + "plain", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _TIER_LABELS_ONLY}, + ["auto_router/complexity_router"], + _CUSTOM_TIERS, + None, + "plain", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _CUSTOM_PROMPT}, + ["auto_router/complexity_router"], + None, + None, + "refused", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _OPERATOR_EXAMPLES}, + [], + _CUSTOM_TIERS, + None, + "refused", + ), + ( + 1, + {"model": "auto_router/complexity_router", "complexity_router_config": _OPERATOR_OPENING_PROMPT}, + [], + _CUSTOM_PROMPT, + None, + "refused", + ), + ( + None, + {"model": "auto_router/complexity_router", "complexity_router_config": _OPERATOR_EXAMPLES}, + ["auto_router/complexity_router"], + _CUSTOM_TIERS, + None, + "plain", + ), ], ) async def test_auto_router_capability_slot_matrix( @@ -6233,10 +6266,16 @@ class TestStrategyRouterWriteValidation: capability = gated_capability_of(effective_params) fake = self._FakeDb(db_models) - live_router = self._live_router_holding_one_capability(limit, config_config) if config_config is not None else None + live_router = ( + self._live_router_holding_one_capability(limit, config_config) if config_config is not None else None + ) with ( - patch("litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: limit), # test-quality-ok: the guard reads the proxy license singleton with no injection seam - patch("litellm.proxy.proxy_server.llm_router", live_router), # test-quality-ok: the guard reads the proxy router global with no injection seam + patch( + "litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: limit + ), # test-quality-ok: the guard reads the proxy license singleton with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", live_router + ), # test-quality-ok: the guard reads the proxy router global with no injection seam patch( # test-quality-ok: the cross-pod publish is the side effect under test; redis is not configured here "litellm.proxy.management_endpoints.model_management_endpoints.publish_config_change", new=AsyncMock(), @@ -6252,7 +6291,9 @@ class TestStrategyRouterWriteValidation: assert capability.subject in str(exc_info.value.detail) assert "'auto_router' feature lifts the limit" in str(exc_info.value.detail) return - async with _auto_router_capability_slot(fake, effective_params=effective_params, model_id=model_id) as tables: + async with _auto_router_capability_slot( + fake, effective_params=effective_params, model_id=model_id + ) as tables: handle = tables if expected == "plain": await handle.create(data={}) @@ -6324,7 +6365,10 @@ class TestStrategyRouterWriteValidation: "_MODELS_ONLY_B": self._MODELS_ONLY_B, } baselines = snapshot_tuning_baselines( - [self._db_router_row(row_id, configs["_TUNED_A" if row_id == "a" else "_TUNED_B"]) for row_id in baseline_rows] + [ + self._db_router_row(row_id, configs["_TUNED_A" if row_id == "a" else "_TUNED_B"]) + for row_id in baseline_rows + ] ) effective_params = { "model": "auto_router/complexity_router", @@ -6347,19 +6391,29 @@ class TestStrategyRouterWriteValidation: ], ) with ( - patch("litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: limit), # test-quality-ok: the guard reads the proxy license singleton with no injection seam - patch("litellm.proxy.proxy_server.llm_router", None), # test-quality-ok: the guard reads the proxy router global with no injection seam - patch("litellm.proxy.proxy_server.heuristic_v1_tuning_baselines", baselines), # test-quality-ok: baselines are a startup-loaded proxy global with no injection seam + patch( + "litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: limit + ), # test-quality-ok: the guard reads the proxy license singleton with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", None + ), # test-quality-ok: the guard reads the proxy router global with no injection seam + patch( + "litellm.proxy.proxy_server.heuristic_v1_tuning_baselines", baselines + ), # test-quality-ok: baselines are a startup-loaded proxy global with no injection seam ): if expected == "refused": with pytest.raises(HTTPException) as exc_info: - async with _auto_router_capability_slot(fake, effective_params=effective_params, model_id=candidate_id): + async with _auto_router_capability_slot( + fake, effective_params=effective_params, model_id=candidate_id + ): pass assert exc_info.value.status_code == 403 assert "changed heuristic scoring rules" in str(exc_info.value.detail) assert "'auto_router' feature lifts the limit" in str(exc_info.value.detail) return - async with _auto_router_capability_slot(fake, effective_params=effective_params, model_id=candidate_id) as table: + async with _auto_router_capability_slot( + fake, effective_params=effective_params, model_id=candidate_id + ) as table: assert hasattr(table, "create") @pytest.mark.asyncio @@ -6386,12 +6440,24 @@ class TestStrategyRouterWriteValidation: ], ) with ( - patch("litellm.proxy.proxy_server.prisma_client", fake), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.llm_router", None), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: 1), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.heuristic_v1_tuning_baselines", baselines), # test-quality-ok: baselines are a startup-loaded proxy global with no injection seam + patch( + "litellm.proxy.proxy_server.prisma_client", fake + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", None + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: 1 + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.heuristic_v1_tuning_baselines", baselines + ), # test-quality-ok: baselines are a startup-loaded proxy global with no injection seam patch( # test-quality-ok: prior auth check needs a live DB; only the tuning quota is under test "litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call", new=AsyncMock(return_value=None), @@ -6423,9 +6489,15 @@ class TestStrategyRouterWriteValidation: fake = self._FakeDb([]) with ( - patch("litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: 1), # test-quality-ok: the guard reads the proxy license singleton with no injection seam - patch("litellm.proxy.proxy_server.llm_router", None), # test-quality-ok: the guard reads the proxy router global with no injection seam - patch("litellm.proxy.proxy_server.heuristic_v1_tuning_baselines", None), # test-quality-ok: baselines are a startup-loaded proxy global with no injection seam + patch( + "litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: 1 + ), # test-quality-ok: the guard reads the proxy license singleton with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", None + ), # test-quality-ok: the guard reads the proxy router global with no injection seam + patch( + "litellm.proxy.proxy_server.heuristic_v1_tuning_baselines", None + ), # test-quality-ok: baselines are a startup-loaded proxy global with no injection seam ): async with _auto_router_capability_slot( fake, @@ -6483,7 +6555,9 @@ class TestStrategyRouterWriteValidation: @pytest.mark.asyncio @pytest.mark.parametrize("config", [_V2, _CAPABILITY, _FUSE]) - async def test_add_new_model_refuses_a_second_gated_classifier_router_before_the_db_write(self, config: Mapping[str, object]) -> None: + async def test_add_new_model_refuses_a_second_gated_classifier_router_before_the_db_write( + self, config: Mapping[str, object] + ) -> None: from litellm.proxy._types import ProxyException from litellm.proxy.management_endpoints.model_management_endpoints import ( add_new_model, @@ -6493,11 +6567,21 @@ class TestStrategyRouterWriteValidation: fake = self._FakeDb(["auto_router/complexity_router"]) with ( - patch("litellm.proxy.proxy_server.prisma_client", fake), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.llm_router", None), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: 1), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.prisma_client", fake + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", None + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: 1 + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam patch( # test-quality-ok: prior auth check needs a live DB; only the license limit is under test "litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call", new=AsyncMock(return_value=None), @@ -6511,7 +6595,9 @@ class TestStrategyRouterWriteValidation: await add_new_model( model_params=Deployment( model_name="second-v2", - litellm_params=LiteLLM_Params(model="auto_router/complexity_router", complexity_router_config=config), + litellm_params=LiteLLM_Params( + model="auto_router/complexity_router", complexity_router_config=config + ), ), user_api_key_dict=admin, ) @@ -6536,10 +6622,18 @@ class TestStrategyRouterWriteValidation: ) fake = self._FakeDb([]) with ( - patch("litellm.proxy.proxy_server.prisma_client", fake), # test-quality-ok: endpoint reads proxy globals with no injection seam - patch("litellm.proxy.proxy_server.llm_router", None), # test-quality-ok: config-based lookup must be absent to drive the stored-row branch - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reaches its DB-write branch only with this process setting - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: authorization branch reads the proxy-wide premium flag + patch( + "litellm.proxy.proxy_server.prisma_client", fake + ), # test-quality-ok: endpoint reads proxy globals with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", None + ), # test-quality-ok: config-based lookup must be absent to drive the stored-row branch + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reaches its DB-write branch only with this process setting + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: authorization branch reads the proxy-wide premium flag patch( # test-quality-ok: inject stored regular row without a database "litellm.proxy.management_endpoints.model_management_endpoints.get_db_model", new=AsyncMock(return_value=regular), @@ -6583,10 +6677,18 @@ class TestStrategyRouterWriteValidation: existing_row.litellm_params = regular.litellm_params.model_dump() fake = self._FakeDb([], existing_row=existing_row) with ( - patch("litellm.proxy.proxy_server.prisma_client", fake), # test-quality-ok: endpoint reads proxy globals with no injection seam - patch("litellm.proxy.proxy_server.llm_router", None), # test-quality-ok: config-based lookup must be absent to drive the stored-row branch - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reaches its DB-write branch only with this process setting - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: authorization branch reads the proxy-wide premium flag + patch( + "litellm.proxy.proxy_server.prisma_client", fake + ), # test-quality-ok: endpoint reads proxy globals with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", None + ), # test-quality-ok: config-based lookup must be absent to drive the stored-row branch + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reaches its DB-write branch only with this process setting + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: authorization branch reads the proxy-wide premium flag patch( # test-quality-ok: endpoint must reject before database authorization needs a live store "litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call", new=AsyncMock(return_value=None), @@ -6609,7 +6711,9 @@ class TestStrategyRouterWriteValidation: @pytest.mark.asyncio @pytest.mark.parametrize("config", [_V2, _CAPABILITY, _FUSE]) - async def test_patch_model_refuses_switching_another_router_to_gated_classifier(self, config: Mapping[str, object]) -> None: + async def test_patch_model_refuses_switching_another_router_to_gated_classifier( + self, config: Mapping[str, object] + ) -> None: """patch_model relays HTTPException as-is, so the license refusal reaches the client as a plain 403.""" from fastapi import HTTPException @@ -6623,11 +6727,21 @@ class TestStrategyRouterWriteValidation: fake = self._FakeDb(["auto_router/complexity_router"]) with ( - patch("litellm.proxy.proxy_server.prisma_client", fake), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.llm_router", None), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: 1), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.prisma_client", fake + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", None + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: 1 + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam patch( # test-quality-ok: the write must be refused before this DB step runs "litellm.proxy.management_endpoints.model_management_endpoints.get_db_model", new=AsyncMock(return_value=self._db_complexity_router(model_id)), @@ -6653,7 +6767,9 @@ class TestStrategyRouterWriteValidation: @pytest.mark.asyncio @pytest.mark.parametrize("config", [_V2, _CAPABILITY, _FUSE]) - async def test_update_model_refuses_switching_another_router_to_gated_classifier(self, config: Mapping[str, object]) -> None: + async def test_update_model_refuses_switching_another_router_to_gated_classifier( + self, config: Mapping[str, object] + ) -> None: from litellm.proxy._types import ProxyException from litellm.proxy.management_endpoints.model_management_endpoints import ( update_model, @@ -6676,11 +6792,21 @@ class TestStrategyRouterWriteValidation: fake = self._FakeDb(["auto_router/complexity_router"], existing_row=existing_row) with ( - patch("litellm.proxy.proxy_server.prisma_client", fake), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.llm_router", None), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: endpoint reads proxy server globals with no injection seam - patch("litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: 1), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.prisma_client", fake + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.llm_router", None + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam + patch( + "litellm.proxy.proxy_server._license_check.auto_router_capability_limit", lambda: 1 + ), # test-quality-ok: endpoint reads proxy server globals with no injection seam patch( # test-quality-ok: prior auth check needs a live DB; only the license limit is under test "litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call", new=AsyncMock(return_value=None), @@ -6770,13 +6896,17 @@ class TestAutoRouterClassifierDefaultPrompt: from litellm.router_strategy.complexity_router import ClassificationRubric, classification_system_prompt for preset in ClassificationRubric: - response = await get_auto_router_classifier_default_prompt(context_window_size=5, classification_rubric=preset) + response = await get_auto_router_classifier_default_prompt( + context_window_size=5, classification_rubric=preset + ) assert response.system_prompt == classification_system_prompt(5, classification_rubric=preset) agentic = await get_auto_router_classifier_default_prompt( context_window_size=5, classification_rubric=ClassificationRubric.AGENTIC ) - chat = await get_auto_router_classifier_default_prompt(context_window_size=5, classification_rubric=ClassificationRubric.CHAT) + chat = await get_auto_router_classifier_default_prompt( + context_window_size=5, classification_rubric=ClassificationRubric.CHAT + ) unset = await get_auto_router_classifier_default_prompt(context_window_size=5) assert "Calibration on engineering tasks" in agentic.system_prompt assert "Calibration on engineering tasks" not in chat.system_prompt @@ -7110,9 +7240,7 @@ class TestEnforceRpmTpmOnModelAdd: class TestBlockModelResponseSerialization: - @pytest.mark.parametrize( - ("route", "blocked"), [("/model/block", True), ("/model/unblock", False)] - ) + @pytest.mark.parametrize(("route", "blocked"), [("/model/block", True), ("/model/unblock", False)]) def test_block_routes_serialize_prisma_row_to_200(self, route, blocked): from datetime import datetime, timezone @@ -7143,13 +7271,19 @@ class TestBlockModelResponseSerialization: app.dependency_overrides[ps.user_api_key_auth] = lambda: admin try: with ( - patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), # test-quality-ok: proxy_server module global is the endpoint's only injection point - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: proxy_server module global is the endpoint's only injection point + patch( + "litellm.proxy.proxy_server.prisma_client", mock_prisma + ), # test-quality-ok: proxy_server module global is the endpoint's only injection point + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: proxy_server module global is the endpoint's only injection point patch( # test-quality-ok: proxy_server module global is the endpoint's only injection point "litellm.proxy.proxy_server.llm_router", MagicMock(**{"get_model_ids.return_value": ["m-block-1"]}), ), - patch("litellm.proxy.proxy_server.redis_usage_cache", None), # test-quality-ok: proxy_server module global is the endpoint's only injection point + patch( + "litellm.proxy.proxy_server.redis_usage_cache", None + ), # test-quality-ok: proxy_server module global is the endpoint's only injection point patch( # test-quality-ok: stubs the cache write so the test observes only response serialization "litellm.proxy.management_endpoints.model_management_endpoints.clear_cache", new=AsyncMock(return_value=ReconcileOutcome(still_desired=None, live_after=None)), @@ -7478,16 +7612,36 @@ class TestTeamMemberAutoRouterWrites: @contextlib.contextmanager def _environment(self, database: MagicMock, row: LiteLLM_ProxyModelTable) -> Iterator[None]: with ( - patch("litellm.proxy.proxy_server.prisma_client", database), # test-quality-ok: [TQ008] endpoint storage singleton injection - patch("litellm.proxy.proxy_server.llm_router", self._catalog()), # test-quality-ok: [TQ008] inject real destination model catalog - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: [TQ008] endpoint storage mode singleton - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: [TQ008] inject licensed process state - patch("litellm.proxy.proxy_server._license_check.auto_router_capability_limit", return_value=None), # test-quality-ok: [TQ008] inject unlimited license result - patch("litellm.proxy.management_endpoints.model_management_endpoints.publish_config_change", new=AsyncMock()), # test-quality-ok: [TQ008] pubsub I/O boundary - patch("litellm.proxy.management_endpoints.model_management_endpoints.create_object_audit_log", new=AsyncMock()), # test-quality-ok: [TQ008] audit database I/O boundary - patch("litellm.proxy.management_endpoints.model_management_endpoints.clear_cache", new=AsyncMock(return_value=ReconcileOutcome( # test-quality-ok: [TQ008] model reload I/O boundary - still_desired=frozenset((row.model_id, "allowed-id")), live_after=frozenset((row.model_id, "allowed-id")) - ))), + patch( + "litellm.proxy.proxy_server.prisma_client", database + ), # test-quality-ok: [TQ008] endpoint storage singleton injection + patch( + "litellm.proxy.proxy_server.llm_router", self._catalog() + ), # test-quality-ok: [TQ008] inject real destination model catalog + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: [TQ008] endpoint storage mode singleton + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: [TQ008] inject licensed process state + patch( + "litellm.proxy.proxy_server._license_check.auto_router_capability_limit", return_value=None + ), # test-quality-ok: [TQ008] inject unlimited license result + patch( + "litellm.proxy.management_endpoints.model_management_endpoints.publish_config_change", new=AsyncMock() + ), # test-quality-ok: [TQ008] pubsub I/O boundary + patch( + "litellm.proxy.management_endpoints.model_management_endpoints.create_object_audit_log", new=AsyncMock() + ), # test-quality-ok: [TQ008] audit database I/O boundary + patch( + "litellm.proxy.management_endpoints.model_management_endpoints.clear_cache", + new=AsyncMock( + return_value=ReconcileOutcome( # test-quality-ok: [TQ008] model reload I/O boundary + still_desired=frozenset((row.model_id, "allowed-id")), + live_after=frozenset((row.model_id, "allowed-id")), + ) + ), + ), ): yield @@ -7548,11 +7702,15 @@ class TestTeamMemberAutoRouterWrites: @staticmethod def _catalog() -> Router: - return Router(model_list=[{ - "model_name": "allowed", - "litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "fake"}, - "model_info": {"id": "allowed-id"}, - }]) + return Router( + model_list=[ + { + "model_name": "allowed", + "litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "fake"}, + "model_info": {"id": "allowed-id"}, + } + ] + ) @pytest.mark.asyncio @pytest.mark.parametrize("endpoint,change", [("patch", "config"), ("legacy", "strategy"), ("patch", "unrelated")]) @@ -7645,9 +7803,7 @@ class TestTeamMemberAutoRouterWrites: @pytest.mark.asyncio @pytest.mark.parametrize("endpoint", ["patch", "legacy"]) @pytest.mark.parametrize("access", ["owner", "peer", "limited-key"]) - async def test_both_update_entries_enforce_creator_and_stamp_member_scope( - self, endpoint: str, access: str - ) -> None: + async def test_both_update_entries_enforce_creator_and_stamp_member_scope(self, endpoint: str, access: str) -> None: from fastapi import HTTPException from litellm.proxy._types import ProxyException @@ -7656,15 +7812,21 @@ class TestTeamMemberAutoRouterWrites: row: Final = self._row() database: Final = self._database(self._team(), row) request: Final = updateDeployment( - litellm_params=updateLiteLLMParams(complexity_router_config={"tiers": {"SIMPLE": "allowed"}, "session_affinity": True}), + litellm_params=updateLiteLLMParams( + complexity_router_config={"tiers": {"SIMPLE": "allowed"}, "session_affinity": True} + ), model_info=ModelInfo(id=row.model_id, team_id="member-team"), ) actor: Final = UserAPIKeyAuth( - user_id="peer" if access == "peer" else "owner", user_role=LitellmUserRoles.INTERNAL_USER, - models=["personal-router"] if access == "limited-key" else ["allowed"], config={"timeout": 60}, + user_id="peer" if access == "peer" else "owner", + user_role=LitellmUserRoles.INTERNAL_USER, + models=["personal-router"] if access == "limited-key" else ["allowed"], + config={"timeout": 60}, ) with self._environment(database, row): - operation: Final = patch_model(row.model_id, request, actor) if endpoint == "patch" else update_model(request, actor) + operation: Final = ( + patch_model(row.model_id, request, actor) if endpoint == "patch" else update_model(request, actor) + ) if access != "owner": with pytest.raises((HTTPException, ProxyException)): await operation @@ -7686,35 +7848,52 @@ class TestTeamMemberAutoRouterWrites: from fastapi import HTTPException from litellm.proxy.management_endpoints.model_management_endpoints import _auto_router_capability_slot - from litellm.proxy.management_helpers.auto_router_permissions import MemberAutoRouterWrite, validate_member_auto_router_config + from litellm.proxy.management_helpers.auto_router_permissions import ( + MemberAutoRouterWrite, + validate_member_auto_router_config, + ) row: Final = self._row() database: Final = self._database(self._team(), row) if changed_state == "revoked": database.transaction.litellm_teamtable.find_unique.return_value = self._team(enabled=False) elif changed_state == "moved": - database.transaction.litellm_proxymodeltable.find_unique.return_value = row.model_copy(update={"model_info": {"team_id": "other-team"}}) + database.transaction.litellm_proxymodeltable.find_unique.return_value = row.model_copy( + update={"model_info": {"team_id": "other-team"}} + ) elif changed_state == "creator": - database.transaction.litellm_proxymodeltable.find_unique.return_value = row.model_copy(update={"created_by": "peer"}) + database.transaction.litellm_proxymodeltable.find_unique.return_value = row.model_copy( + update={"created_by": "peer"} + ) elif changed_state == "collision": database.transaction.litellm_proxymodeltable.find_many.return_value = [row] config: Final = validate_member_auto_router_config({"tiers": {"SIMPLE": "allowed"}}) grant: Final = MemberAutoRouterWrite( actor=UserAPIKeyAuth(user_id="owner", user_role=LitellmUserRoles.INTERNAL_USER, models=["allowed"]), - team_id="member-team", model_id=None if changed_state in ("collision", "global-alias") else row.model_id, - public_name="personal-router", updated_at=None, config=config, default_model="allowed", + team_id="member-team", + model_id=None if changed_state in ("collision", "global-alias") else row.model_id, + public_name="personal-router", + updated_at=None, + config=config, + default_model="allowed", ) with ( self._environment(database, row), - patch("litellm.model_alias_map", {"personal-router": "allowed"} if changed_state == "global-alias" else {}), # test-quality-ok: [TQ008] inject alias namespace for collision behavior + patch( + "litellm.model_alias_map", {"personal-router": "allowed"} if changed_state == "global-alias" else {} + ), # test-quality-ok: [TQ008] inject alias namespace for collision behavior ): if changed_state != "allowed": with pytest.raises(HTTPException) as denied: - async with _auto_router_capability_slot(database, effective_params={}, model_id=grant.model_id, member_write=grant): + async with _auto_router_capability_slot( + database, effective_params={}, model_id=grant.model_id, member_write=grant + ): pytest.fail("An invalidated grant reached the database writer") assert denied.value.status_code == (409 if changed_state in ("collision", "global-alias") else 403) return - async with _auto_router_capability_slot(database, effective_params={}, model_id=grant.model_id, member_write=grant) as table: + async with _auto_router_capability_slot( + database, effective_params={}, model_id=grant.model_id, member_write=grant + ) as table: await table.update(where={"model_id": row.model_id}, data={"updated_by": "owner"}) assert database.transaction.query_raw.await_count == 2 database.transaction.litellm_proxymodeltable.update.assert_awaited_once() @@ -7728,20 +7907,32 @@ class TestTeamMemberAutoRouterWrites: row: Final = self._row() database: Final = self._database(self._team(enabled=access != "opt-out"), row) actor: Final = UserAPIKeyAuth( - user_id="owner", user_role=LitellmUserRoles.INTERNAL_USER, - models=["personal-router"] if access == "limited-key" else ["allowed"], config={"timeout": 60}, + user_id="owner", + user_role=LitellmUserRoles.INTERNAL_USER, + models=["personal-router"] if access == "limited-key" else ["allowed"], + config={"timeout": 60}, ) deployment: Final = Deployment( model_name="new-personal-router", - litellm_params=LiteLLM_Params(model="auto_router/complexity_router", complexity_router_config={"tiers": {"SIMPLE": "allowed"}}), + litellm_params=LiteLLM_Params( + model="auto_router/complexity_router", complexity_router_config={"tiers": {"SIMPLE": "allowed"}} + ), model_info=ModelInfo(id=row.model_id, team_id="member-team"), ) with ( self._environment(database, row), - patch("litellm.proxy.proxy_server.proxy_config.add_deployment", new=AsyncMock(return_value=ReconcileOutcome( # test-quality-ok: [TQ008] model reload I/O boundary - still_desired=frozenset((row.model_id, "allowed-id")), live_after=frozenset((row.model_id, "allowed-id")) - ))), - patch("litellm.proxy.management_endpoints.model_management_endpoints.append_team_models", new=AsyncMock()) as appended, # test-quality-ok: [TQ008] persistence boundary; the appended scope is asserted + patch( + "litellm.proxy.proxy_server.proxy_config.add_deployment", + new=AsyncMock( + return_value=ReconcileOutcome( # test-quality-ok: [TQ008] model reload I/O boundary + still_desired=frozenset((row.model_id, "allowed-id")), + live_after=frozenset((row.model_id, "allowed-id")), + ) + ), + ), + patch( + "litellm.proxy.management_endpoints.model_management_endpoints.append_team_models", new=AsyncMock() + ) as appended, # test-quality-ok: [TQ008] persistence boundary; the appended scope is asserted ): if access != "allowed": with pytest.raises(ProxyException) as denied: @@ -7772,10 +7963,18 @@ class TestModelManagementActorEdges: model_info=ModelInfo(id="internal-model-id"), ) with ( - patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.general_settings", {}), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.prisma_client", prisma + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.general_settings", {} + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam ): with pytest.raises(ProxyException) as exc_info: await add_new_model(model_params=deployment, user_api_key_dict=actor) @@ -7789,9 +7988,7 @@ class TestModelManagementActorEdges: from litellm.proxy._types import ProxyException from litellm.proxy.management_endpoints.model_management_endpoints import add_new_model - actor: Final = UserAPIKeyAuth( - user_id="view-only-user", user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY - ) + actor: Final = UserAPIKeyAuth(user_id="view-only-user", user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY) prisma: Final = MagicMock() deployment: Final = Deployment( model_name="view-only-model", @@ -7799,10 +7996,18 @@ class TestModelManagementActorEdges: model_info=ModelInfo(id="view-only-model-id"), ) with ( - patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.general_settings", {}), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.prisma_client", prisma + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.general_settings", {} + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam ): with pytest.raises(ProxyException) as exc_info: await add_new_model(model_params=deployment, user_api_key_dict=actor) @@ -7824,10 +8029,18 @@ class TestModelManagementActorEdges: model_info=ModelInfo(id="database-disabled-model-id"), ) with ( - patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.store_model_in_db", False), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.general_settings", {}), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.prisma_client", prisma + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", False + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.general_settings", {} + ), # test-quality-ok: [TQ008] endpoint reads proxy-server state through its only test seam ): with pytest.raises(ProxyException) as exc_info: await add_new_model(model_params=deployment, user_api_key_dict=actor) @@ -7858,10 +8071,18 @@ class TestModelManagementActorEdges: router.get_model_ids.return_value = [model_id] actor: Final = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) with ( - patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.llm_router", router), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.prisma_client", prisma + ), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.llm_router", router + ), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam patch( # test-quality-ok: [TQ008] isolate persistence from encryption implementation "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", side_effect=lambda value: value, @@ -7907,10 +8128,18 @@ class TestModelManagementActorEdges: router.get_model_ids.return_value = [model_id] actor: Final = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) with ( - patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.llm_router", router), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.prisma_client", prisma + ), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.llm_router", router + ), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: [TQ008] update endpoint reads proxy-server state through its only test seam patch( # test-quality-ok: [TQ008] isolate persistence from encryption implementation "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", side_effect=lambda value: value, @@ -7950,10 +8179,18 @@ class TestModelManagementActorEdges: ) actor: Final = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) with ( - patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: [TQ008] patch endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.llm_router", router), # test-quality-ok: [TQ008] patch endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: [TQ008] patch endpoint reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: [TQ008] patch endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.prisma_client", prisma + ), # test-quality-ok: [TQ008] patch endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.llm_router", router + ), # test-quality-ok: [TQ008] patch endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: [TQ008] patch endpoint reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: [TQ008] patch endpoint reads proxy-server state through its only test seam ): with pytest.raises(ProxyException) as exc_info: await patch_model( @@ -7984,10 +8221,18 @@ class TestModelManagementActorEdges: actor: Final = UserAPIKeyAuth(user_id="internal-user", user_role=LitellmUserRoles.INTERNAL_USER) prisma: Final = MagicMock() with ( - patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.general_settings", {}), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.prisma_client", prisma + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.general_settings", {} + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam self._client_for(actor) as client, ): response: Final = client.post( @@ -8028,10 +8273,18 @@ class TestModelManagementActorEdges: router.get_model_ids.return_value = [model_id] actor: Final = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) with ( - patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.llm_router", router), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.prisma_client", prisma + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.llm_router", router + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam patch( # test-quality-ok: [TQ008] isolate persistence from encryption implementation "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", side_effect=lambda value: value, @@ -8073,10 +8326,18 @@ class TestModelManagementActorEdges: ) actor: Final = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) with ( - patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.llm_router", router), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam - patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.prisma_client", prisma + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.llm_router", router + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.store_model_in_db", True + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam + patch( + "litellm.proxy.proxy_server.premium_user", True + ), # test-quality-ok: [TQ008] route reads proxy-server state through its only test seam self._client_for(actor) as client, ): response: Final = client.patch(