test(team): prove default access groups reach the saved team row and type the settings fixtures

litellm/proxy/ui_crud_endpoints/proxy_setting_endpoints.py

@@ -1172,12 +1172,6 @@ async def update_internal_user_settings(
 
 
 async def _validate_default_access_groups_exist(access_group_ids: Sequence[str]) -> None:
-    """Reject default access groups that cannot be attached.
-
-    Teams are created from these settings long after they are saved, and /team/new does
-    not check that an access group exists, so an unknown id would silently attach nothing
-    to every future team instead of failing here, where the admin can still fix it.
-    """
     if not access_group_ids:
         return
 

tests/unit/proxy/management_endpoints/test_team_default_params.py

@@ -3,6 +3,7 @@ Tests for applying default team params during team creation
 and loading default_team_params from DB on startup.
 """
 
+from typing import Final
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -11,6 +12,7 @@ from fastapi import HTTPException
 
 import litellm
 from litellm.proxy._types import (
+    LiteLLM_TeamTable,
     LiteLLM_BudgetTable,
     LiteLLM_OrganizationTable,
     NewTeamRequest,
@@ -214,39 +216,58 @@ class TestNewTeamDefaultParamsApplied:
         assert data.rpm_limit == 500
         assert data.team_member_permissions == ["/key/generate", "/key/update"]
 
-    @pytest.mark.asyncio
-    async def test_default_access_group_ids_attach_to_a_team_created_without_any(self, monkeypatch):
-        """A central access group configured in default_team_params lands on every
-        /team/new call that omits access_group_ids, not only on SSO-created teams."""
+    async def _create_team(
+        self, monkeypatch: pytest.MonkeyPatch, data: NewTeamRequest
+    ) -> tuple[dict[str, object], AsyncMock, AsyncMock]:
+        from litellm.proxy import proxy_server
+        from litellm.proxy.management_endpoints import team_endpoints
         from litellm.proxy.management_endpoints.team_endpoints import new_team
 
+        prisma_client: Final = proxy_server.prisma_client
+        prisma_client.jsonify_team_object = MagicMock(side_effect=lambda db_data: db_data)
+
+        async def _create(data: dict[str, object], include: object) -> LiteLLM_TeamTable:
+            return LiteLLM_TeamTable(**data)
+
+        create: Final = AsyncMock(side_effect=_create)
+        tx: Final = MagicMock()
+        tx.litellm_teamtable.create = create
+        prisma_client.db.tx.return_value.__aenter__.return_value = tx
+        reconcile: Final = AsyncMock(return_value=())
+        monkeypatch.setattr(team_endpoints, "reconcile_team_access_group_membership", reconcile)
+        monkeypatch.setattr(team_endpoints, "invalidate_access_group_caches", AsyncMock())
+        monkeypatch.setattr(team_endpoints, "_add_team_members_to_team", AsyncMock())
+
+        response: Final = await new_team(data=data, user_api_key_dict=self._make_admin_auth(), http_request=MagicMock())
+        return response, create, reconcile
+
+    @pytest.mark.asyncio
+    async def test_default_access_group_ids_attach_to_a_team_created_without_any(self, monkeypatch: pytest.MonkeyPatch):
         monkeypatch.setattr(litellm, "default_team_params", {"access_group_ids": ["central-models"]})
 
-        data = NewTeamRequest(team_alias="my-team")
+        response, create, reconcile = await self._create_team(monkeypatch, NewTeamRequest(team_alias="my-team"))
 
-        try:
-            await new_team(data=data, user_api_key_dict=self._make_admin_auth(), http_request=MagicMock())
-        except Exception:
-            pass
-
-        assert data.access_group_ids == ["central-models"]
+        assert create.await_args is not None
+        assert create.await_args.kwargs["data"]["access_group_ids"] == ["central-models"]
+        assert response["access_group_ids"] == ["central-models"]
+        reconcile.assert_awaited_once()
+        assert reconcile.await_args is not None
+        assert reconcile.await_args.args[1] == response["team_id"]
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("explicit_access_group_ids", [["research"], []])
-    async def test_explicit_access_group_ids_win_over_the_default(self, monkeypatch, explicit_access_group_ids):
-        """An explicit list, including an empty one, is the caller's choice and is kept."""
-        from litellm.proxy.management_endpoints.team_endpoints import new_team
-
+    async def test_explicit_access_group_ids_win_over_the_default(
+        self, monkeypatch: pytest.MonkeyPatch, explicit_access_group_ids: list[str]
+    ):
         monkeypatch.setattr(litellm, "default_team_params", {"access_group_ids": ["central-models"]})
 
-        data = NewTeamRequest(team_alias="my-team", access_group_ids=explicit_access_group_ids)
+        response, create, _ = await self._create_team(
+            monkeypatch, NewTeamRequest(team_alias="my-team", access_group_ids=explicit_access_group_ids)
+        )
 
-        try:
-            await new_team(data=data, user_api_key_dict=self._make_admin_auth(), http_request=MagicMock())
-        except Exception:
-            pass
-
-        assert data.access_group_ids == explicit_access_group_ids
+        assert create.await_args is not None
+        assert create.await_args.kwargs["data"]["access_group_ids"] == explicit_access_group_ids
+        assert response["access_group_ids"] == explicit_access_group_ids
 
     @pytest.mark.asyncio
     async def test_explicit_values_not_overridden(self, monkeypatch):

tests/unit/proxy/ui_crud_endpoints/test_proxy_setting_endpoints.py

@@ -1,8 +1,8 @@
 import json
 import os
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from typing import Final
-from unittest.mock import patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from fastapi.testclient import TestClient
@@ -3459,39 +3459,40 @@ def test_update_default_team_settings_without_organization_skips_lookup(
 
 
 @pytest.fixture
-def mock_access_group_lookup(monkeypatch):
-    """Fake the access group table so the default team settings endpoint can check ids."""
-    from unittest.mock import AsyncMock, MagicMock
-
+def mock_access_group_lookup(monkeypatch: pytest.MonkeyPatch) -> Callable[[frozenset[str]], AsyncMock]:
     import litellm
     import litellm.proxy.proxy_server as proxy_server_module
 
-    existing_access_group_ids: set = set()
+    def _install(existing_access_group_ids: frozenset[str]) -> AsyncMock:
+        async def _find_many(where: dict[str, dict[str, list[str]]]) -> list[MagicMock]:
+            return [
+                MagicMock(access_group_id=access_group_id)
+                for access_group_id in where["access_group_id"]["in"]
+                if access_group_id in existing_access_group_ids
+            ]
 
-    async def _find_many(where):
-        return [
-            MagicMock(access_group_id=access_group_id)
-            for access_group_id in where["access_group_id"]["in"]
-            if access_group_id in existing_access_group_ids
-        ]
+        find_many: Final = AsyncMock(side_effect=_find_many)
+        fake_prisma: Final = MagicMock()
+        fake_prisma.db.litellm_accessgrouptable.find_many = find_many
 
-    find_many = AsyncMock(side_effect=_find_many)
-    fake_prisma = MagicMock()
-    fake_prisma.db.litellm_accessgrouptable.find_many = find_many
+        monkeypatch.setattr(proxy_server_module, "prisma_client", fake_prisma)
+        monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", True)
+        monkeypatch.setattr(litellm, "default_team_params", {})
+        return find_many
 
-    monkeypatch.setattr(proxy_server_module, "prisma_client", fake_prisma)
-    monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", True)
-    monkeypatch.setattr(litellm, "default_team_params", {})
-
-    return {"existing_access_group_ids": existing_access_group_ids, "find_many": find_many}
+    return _install
 
 
-def test_update_default_team_settings_rejects_unknown_access_group(mock_proxy_config, mock_auth, mock_access_group_lookup):
-    """/team/new never checks that an access group exists, so an unknown default id
-    would silently attach nothing to every future team. Catch it where the admin typed it."""
-    mock_access_group_lookup["existing_access_group_ids"].add("real-group")
+def test_update_default_team_settings_rejects_unknown_access_group(
+    mock_proxy_config: dict[str, object],
+    mock_auth: None,
+    mock_access_group_lookup: Callable[[frozenset[str]], AsyncMock],
+) -> None:
+    import litellm
 
-    resp = client.patch(
+    mock_access_group_lookup(frozenset({"real-group"}))
+
+    resp: Final = client.patch(
         "/update/default_team_settings",
         json={"access_group_ids": ["real-group", "ghost-group"]},
     )
@@ -3500,17 +3501,19 @@ def test_update_default_team_settings_rejects_unknown_access_group(mock_proxy_co
     assert "ghost-group" in resp.json()["detail"]["error"]
     assert "real-group" not in resp.json()["detail"]["error"]
     assert mock_proxy_config["save_call_count"]() == 0
-
-    import litellm
-
     assert litellm.default_team_params == {}
 
 
-def test_update_default_team_settings_saves_existing_access_groups(mock_proxy_config, mock_auth, mock_access_group_lookup):
-    """Real access group ids save and reach the in-memory settings /team/new reads."""
-    mock_access_group_lookup["existing_access_group_ids"].update({"real-group", "other-group"})
+def test_update_default_team_settings_saves_existing_access_groups(
+    mock_proxy_config: dict[str, object],
+    mock_auth: None,
+    mock_access_group_lookup: Callable[[frozenset[str]], AsyncMock],
+) -> None:
+    import litellm
 
-    resp = client.patch(
+    mock_access_group_lookup(frozenset({"real-group", "other-group"}))
+
+    resp: Final = client.patch(
         "/update/default_team_settings",
         json={"access_group_ids": ["real-group", "other-group"]},
     )
@@ -3518,23 +3521,23 @@ def test_update_default_team_settings_saves_existing_access_groups(mock_proxy_co
     assert resp.status_code == 200, resp.text
     assert resp.json()["settings"]["access_group_ids"] == ["real-group", "other-group"]
     assert mock_proxy_config["save_call_count"]() == 1
-
-    import litellm
-
     assert litellm.default_team_params["access_group_ids"] == ["real-group", "other-group"]
 
 
 def test_update_default_team_settings_with_empty_access_groups_skips_lookup(
-    mock_proxy_config, mock_auth, mock_access_group_lookup
-):
-    """Clearing the default access groups must not pay for a DB round trip."""
-    resp = client.patch(
+    mock_proxy_config: dict[str, object],
+    mock_auth: None,
+    mock_access_group_lookup: Callable[[frozenset[str]], AsyncMock],
+) -> None:
+    find_many: Final = mock_access_group_lookup(frozenset())
+
+    resp: Final = client.patch(
         "/update/default_team_settings",
         json={"access_group_ids": []},
     )
 
     assert resp.status_code == 200, resp.text
-    mock_access_group_lookup["find_many"].assert_not_awaited()
+    find_many.assert_not_awaited()
     assert mock_proxy_config["save_call_count"]() == 1