diff --git a/litellm/proxy/management_endpoints/scim/scim_v2.py b/litellm/proxy/management_endpoints/scim/scim_v2.py index d1d1c5959d8..525ce9f0e89 100644 --- a/litellm/proxy/management_endpoints/scim/scim_v2.py +++ b/litellm/proxy/management_endpoints/scim/scim_v2.py @@ -1932,8 +1932,16 @@ async def delete_group( async def _process_group_patch_operations( patch_ops: SCIMPatchOp, existing_team, prisma_client -) -> Tuple[Dict[str, Any], Set[str]]: - """Process patch operations for a group and return update data and final members.""" +) -> Tuple[Dict[str, Any], Set[str], Set[str] | None]: + """Process patch operations for a group and return update data, final members + and, when the request contained a member ``replace`` op, the absolute target + roster it declared (``None`` otherwise). + + ``add``/``remove`` are deltas relative to the current roster, but ``replace`` + is absolute: it declares the roster is exactly this set, so the caller must + reconcile against it as a set-to-target rather than rebasing it onto a + concurrently-mutated roster. + """ update_data: Dict[str, Any] = {} # Create a fresh copy of existing metadata to avoid Prisma issues @@ -2019,7 +2027,12 @@ async def _process_group_patch_operations( if metadata: update_data["metadata"] = metadata - return update_data, final_members + member_replace_present = any( + op.op == "replace" and (op.path or "").lower().startswith("members") for op in patch_ops.Operations + ) + replace_target = set(final_members) if member_replace_present else None + + return update_data, final_members, replace_target async def _apply_group_patch_updates(group_id: str, update_data: Dict[str, Any], prisma_client): @@ -2090,27 +2103,29 @@ async def patch_group( existing_team = await _check_team_exists(group_id) # Process patch operations - update_data, final_members = await _process_group_patch_operations(patch_ops, existing_team, prisma_client) + update_data, final_members, replace_target = await _process_group_patch_operations( + patch_ops, existing_team, prisma_client + ) - # Track current members BEFORE update for comparison - current_members = set(await _get_team_member_user_ids_from_team(existing_team)) + snapshot_members = set(await _get_team_member_user_ids_from_team(existing_team)) + intended_add = final_members - snapshot_members + intended_remove = snapshot_members - final_members # Apply the metadata/displayName updates to the database updated_team = await _apply_group_patch_updates(group_id, update_data, prisma_client) - # Refresh team data from database to get the latest state after concurrent updates - # This prevents race conditions when multiple PATCH requests come in simultaneously refreshed_team = await TeamRepository(prisma_client).table.find_unique(where={"team_id": group_id}) - if refreshed_team: - # Re-read current members from refreshed team to account for concurrent updates - refreshed_current_members = set( - await _get_team_member_user_ids_from_team(LiteLLM_TeamTable(**refreshed_team.model_dump())) - ) - # Use the refreshed members for comparison - current_members = refreshed_current_members + refreshed_current = ( + set(await _get_team_member_user_ids_from_team(LiteLLM_TeamTable(**refreshed_team.model_dump()))) + if refreshed_team + else snapshot_members + ) - # Handle user-team relationship changes - await _handle_group_membership_changes(group_id, current_members, final_members) + effective_final = ( + replace_target if replace_target is not None else (refreshed_current | intended_add) - intended_remove + ) + + await _handle_group_membership_changes(group_id, refreshed_current, effective_final) # A rename can flip whether this group matches scim_admin_group by display # name, so retained members must be re-resolved too, not just the ones whose @@ -2119,7 +2134,7 @@ async def patch_group( alias_changed = new_alias != existing_team.team_alias await _recompute_scim_member_roles( prisma_client, - (current_members | final_members if alias_changed else current_members ^ final_members), + (refreshed_current | effective_final if alias_changed else refreshed_current ^ effective_final), ) # Refresh team one more time to get final state after membership changes diff --git a/litellm/proxy/management_endpoints/team_endpoints.py b/litellm/proxy/management_endpoints/team_endpoints.py index 70c002d2d2d..3a7e89aa20d 100644 --- a/litellm/proxy/management_endpoints/team_endpoints.py +++ b/litellm/proxy/management_endpoints/team_endpoints.py @@ -2375,7 +2375,15 @@ async def _add_team_members_to_team( user_api_key_dict: UserAPIKeyAuth, litellm_proxy_admin_name: str, ) -> Tuple[LiteLLM_TeamTable, List[LiteLLM_UserTable], List[LiteLLM_TeamMembership]]: - """Add team members to the team.""" + """Add team members to the team. + + The members_with_roles reconciliation runs inside a transaction that locks + the team row with ``SELECT ... FOR UPDATE`` before reading the current + membership. Concurrent /team/member_add calls for the same team therefore + serialize on the row lock and each appends onto the other's committed + result, instead of both rewriting the whole JSON array from a stale + snapshot (which silently drops one member on the losing write). + """ # Process and add new members updated_users, updated_team_memberships = await _process_team_members( data=data, @@ -2385,19 +2393,22 @@ async def _add_team_members_to_team( litellm_proxy_admin_name=litellm_proxy_admin_name, ) - # Update team members list - await _update_team_members_list( - data=data, - complete_team_data=complete_team_data, - updated_users=updated_users, - ) + async with prisma_client.tx() as tx: + complete_team_data.members_with_roles = await TeamRepository(prisma_client).get_members_with_roles_locked( + tx, data.team_id + ) - # ADD MEMBER TO TEAM - _db_team_members = [m.model_dump() for m in complete_team_data.members_with_roles] - updated_team = await TeamRepository(prisma_client).table.update( - where={"team_id": data.team_id}, - data={"members_with_roles": json.dumps(_db_team_members)}, # type: ignore - ) + await _update_team_members_list( + data=data, + complete_team_data=complete_team_data, + updated_users=updated_users, + ) + + _db_team_members = [m.model_dump() for m in complete_team_data.members_with_roles] + updated_team = await tx.litellm_teamtable.update( + where={"team_id": data.team_id}, + data={"members_with_roles": json.dumps(_db_team_members)}, + ) return updated_team, updated_users, updated_team_memberships diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 43921a847a9..1dbc0ad1837 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -172,6 +172,7 @@ from litellm.types.utils import LLMResponseTypes, LoggedLiteLLMParams if TYPE_CHECKING: from opentelemetry.trace import Span as _Span + from prisma.client import TransactionManager from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj @@ -2922,6 +2923,14 @@ class PrismaClient: return self.db.writer return self.db + def tx(self) -> "TransactionManager": + """Open an interactive transaction on the writer. + + Callers go through this instead of reaching into ``self.db`` so writer + selection and read-replica routing stay encapsulated in the wrapper. + """ + return cast("TransactionManager", self.db.tx()) # cast-ok: wrappers delegate tx via __getattr__ (untyped) + def get_request_status(self, payload: Union[dict, SpendLogsPayload]) -> Literal["success", "failure"]: """ Determine if a request was successful or failed based on payload metadata. diff --git a/litellm/repositories/team_repository.py b/litellm/repositories/team_repository.py index 3227aa812ca..68875bd7972 100644 --- a/litellm/repositories/team_repository.py +++ b/litellm/repositories/team_repository.py @@ -4,11 +4,18 @@ Team repository for database operations on LiteLLM_TeamTable. import json from datetime import datetime -from typing import Any, Dict, List, Optional, Type +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Type -from litellm.models.team import LiteLLM_TeamTable +from pydantic import TypeAdapter + +from litellm.models.team import LiteLLM_TeamTable, Member from litellm.repositories.base_repository import BaseRepository +if TYPE_CHECKING: + from prisma import Prisma + +_MEMBERS_WITH_ROLES_ADAPTER = TypeAdapter(list[Member]) + class TeamRepository(BaseRepository[LiteLLM_TeamTable]): """Repository for team database operations.""" @@ -46,6 +53,24 @@ class TeamRepository(BaseRepository[LiteLLM_TeamTable]): return LiteLLM_TeamTable(**data) + async def get_members_with_roles_locked(self, tx: "Prisma", team_id: str) -> List[Member]: + """Return the team's members_with_roles, locking the row FOR UPDATE. + + Must be called inside a transaction so the row lock is held until + commit. This serializes concurrent membership writers on the team row + so the losing writer appends onto the winner's committed result instead + of overwriting it from a stale snapshot. + """ + rows = await tx.query_raw( + 'SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = $1 FOR UPDATE', + team_id, + ) + raw_value = rows[0]["members_with_roles"] if rows else None + parsed = json.loads(raw_value) if isinstance(raw_value, str) else raw_value + if not parsed: + return [] + return _MEMBERS_WITH_ROLES_ADAPTER.validate_python(parsed) + async def find_by_id(self, team_id: str, id_field: str = "team_id") -> Optional[LiteLLM_TeamTable]: return await super().find_by_id(team_id, id_field) diff --git a/tests/proxy_unit_tests/test_proxy_server.py b/tests/proxy_unit_tests/test_proxy_server.py index 212f7772cad..bedd4dd1838 100644 --- a/tests/proxy_unit_tests/test_proxy_server.py +++ b/tests/proxy_unit_tests/test_proxy_server.py @@ -1252,6 +1252,17 @@ async def test_create_team_member_add(prisma_client, new_member_method): return_value=LiteLLM_TeamTableCachedObj(team_id="1234") ) + tx_mock = AsyncMock() + tx_mock.query_raw = AsyncMock(return_value=[{"members_with_roles": []}]) + tx_mock.litellm_teamtable = team_mock_client + tx_cm = MagicMock() + tx_cm.__aenter__ = AsyncMock(return_value=tx_mock) + tx_cm.__aexit__ = AsyncMock(return_value=None) + original_tx = litellm.proxy.proxy_server.prisma_client.tx + litellm.proxy.proxy_server.prisma_client.tx = MagicMock( + return_value=tx_cm + ) + print(f"team_member_add_request={team_member_add_request}") await team_member_add( data=team_member_add_request, @@ -1273,6 +1284,7 @@ async def test_create_team_member_add(prisma_client, new_member_method): ) litellm.proxy.proxy_server.prisma_client.db.litellm_teamtable = original_val + litellm.proxy.proxy_server.prisma_client.tx = original_tx @pytest.mark.parametrize("team_member_role", ["admin", "user"]) @@ -1434,42 +1446,51 @@ async def test_create_team_member_add_team_admin( mock_litellm_usertable.find_unique = AsyncMock(return_value=None) team_mock_client = AsyncMock() - original_val = getattr( - litellm.proxy.proxy_server.prisma_client.db, "litellm_teamtable" - ) - litellm.proxy.proxy_server.prisma_client.db.litellm_teamtable = team_mock_client - team_mock_client.update = AsyncMock( return_value=LiteLLM_TeamTableCachedObj(team_id="1234") ) - try: - await team_member_add( - data=team_member_add_request, - user_api_key_dict=valid_token, + tx_mock = AsyncMock() + tx_mock.query_raw = AsyncMock(return_value=[{"members_with_roles": []}]) + tx_mock.litellm_teamtable = team_mock_client + tx_cm = MagicMock() + tx_cm.__aenter__ = AsyncMock(return_value=tx_mock) + tx_cm.__aexit__ = AsyncMock(return_value=None) + + with ( + patch.object( + litellm.proxy.proxy_server.prisma_client.db, + "litellm_teamtable", + team_mock_client, + ), + patch.object( + litellm.proxy.proxy_server.prisma_client, + "tx", + MagicMock(return_value=tx_cm), + ), + ): + try: + await team_member_add( + data=team_member_add_request, + user_api_key_dict=valid_token, + ) + except HTTPException as e: + if user_role == "user": + assert e.status_code == 403 + return + else: + raise e + + mock_client.assert_called() + + assert ( + mock_client.call_args.kwargs["data"]["create"]["max_budget"] + == litellm.max_internal_user_budget + ) + assert ( + mock_client.call_args.kwargs["data"]["create"]["budget_duration"] + == litellm.internal_user_budget_duration ) - except HTTPException as e: - if user_role == "user": - assert e.status_code == 403 - return - else: - raise e - - mock_client.assert_called() - - print(f"mock_client.call_args: {mock_client.call_args}") - print("mock_client.call_args.kwargs: {}".format(mock_client.call_args.kwargs)) - - assert ( - mock_client.call_args.kwargs["data"]["create"]["max_budget"] - == litellm.max_internal_user_budget - ) - assert ( - mock_client.call_args.kwargs["data"]["create"]["budget_duration"] - == litellm.internal_user_budget_duration - ) - - litellm.proxy.proxy_server.prisma_client.db.litellm_teamtable = original_val @pytest.mark.asyncio diff --git a/tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py b/tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py index c31029eb54d..850b8fc7cfd 100644 --- a/tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py @@ -1909,7 +1909,7 @@ async def test_process_group_patch_operations_with_flag_true_creates_users(mocke ) # Execute the function - update_data, final_members = await _process_group_patch_operations( + update_data, final_members, _ = await _process_group_patch_operations( patch_ops=patch_ops, existing_team=mock_existing_team, prisma_client=mock_prisma_client, @@ -2948,7 +2948,7 @@ async def test_process_group_patch_operations_add_retains_existing_members( return_value=mocker.MagicMock(user_id="new-user") ) - _, final_members = await _process_group_patch_operations( + _, final_members, _ = await _process_group_patch_operations( patch_ops=patch_ops, existing_team=existing_team, prisma_client=mock_prisma_client, @@ -2996,7 +2996,7 @@ async def test_process_group_patch_operations_remove_uses_members_with_roles( return_value=mocker.MagicMock(user_id="drop-user") ) - _, final_members = await _process_group_patch_operations( + _, final_members, _ = await _process_group_patch_operations( patch_ops=patch_ops, existing_team=existing_team, prisma_client=mock_prisma_client, @@ -3186,3 +3186,194 @@ async def test_delete_user_skips_teams_where_not_a_member(mocker): team_member_delete_mock.assert_not_awaited() mock_prisma_client.db.litellm_usertable.delete.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_patch_group_add_applies_delta_and_keeps_concurrent_add(mocker): + """A group PATCH op:add must be applied as a delta against the live roster, + not as a snapshot-based absolute target. + + When a concurrent PATCH has already added a member between this request's + initial read and its post-write refresh, that member shows up in the + refreshed roster but not in this request's snapshot-derived target. Diffing + the refreshed roster against the snapshot target would issue a spurious + team_member_delete for the concurrently-added member. Applying only this + request's intended delta on top of the refreshed roster must retain them. + """ + from litellm.proxy.management_endpoints.scim.scim_transformations import ( + ScimTransformations, + ) + + group_id = "team-concurrent" + + snapshot_team = LiteLLM_TeamTable( + team_id=group_id, + team_alias="Group", + members_with_roles=[Member(user_id="zed", role="user")], + metadata={"externalId": "grp-ext"}, + ) + refreshed_team = LiteLLM_TeamTable( + team_id=group_id, + team_alias="Group", + members_with_roles=[ + Member(user_id="zed", role="user"), + Member(user_id="alice", role="user"), + ], + metadata={"externalId": "grp-ext"}, + ) + final_team = LiteLLM_TeamTable( + team_id=group_id, + team_alias="Group", + members_with_roles=[ + Member(user_id="zed", role="user"), + Member(user_id="alice", role="user"), + Member(user_id="bob", role="user"), + ], + metadata={"externalId": "grp-ext"}, + ) + + patch_ops = SCIMPatchOp( + schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + Operations=[SCIMPatchOperation(op="add", path="members", value=[{"value": "bob"}])], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock( + side_effect=[snapshot_team, refreshed_team, final_team] + ) + mock_prisma_client.db.litellm_teamtable.update = AsyncMock(return_value=final_team) + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=mocker.MagicMock()) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + patch_membership_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.patch_team_membership", + AsyncMock(), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles", + AsyncMock(), + ) + mocker.patch.object( + ScimTransformations, + "transform_litellm_team_to_scim_group", + AsyncMock( + return_value=SCIMGroup( + schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"], + id=group_id, + displayName="Group", + ) + ), + ) + + await patch_group(group_id=group_id, patch_ops=patch_ops) + + calls = patch_membership_mock.call_args_list + + removed_user_ids = { + call.kwargs["user_id"] for call in calls if call.kwargs.get("teams_ids_to_remove_user_from") == [group_id] + } + assert removed_user_ids == set() + + added_user_ids = { + call.kwargs["user_id"] for call in calls if call.kwargs.get("teams_ids_to_add_user_to") == [group_id] + } + assert added_user_ids == {"bob"} + + +@pytest.mark.asyncio +async def test_patch_group_replace_stays_absolute_against_concurrent_roster(mocker): + """A group PATCH ``replace`` op declares the roster is exactly the given set, + so it must reconcile as a set-to-target, not as a delta. + + Unlike ``add``/``remove``, ``replace`` is absolute. A member that another + request added concurrently is present in the refreshed roster but not in the + replace target, and ``replace`` must drop it. Rebasing the replace onto the + refreshed roster (the delta behavior correct only for add/remove) would + wrongly retain that concurrently-added member. + """ + from litellm.proxy.management_endpoints.scim.scim_transformations import ( + ScimTransformations, + ) + + group_id = "team-replace-concurrent" + + snapshot_team = LiteLLM_TeamTable( + team_id=group_id, + team_alias="Group", + members_with_roles=[Member(user_id="zed", role="user")], + metadata={"externalId": "grp-ext"}, + ) + refreshed_team = LiteLLM_TeamTable( + team_id=group_id, + team_alias="Group", + members_with_roles=[ + Member(user_id="alice", role="user"), + Member(user_id="bob", role="user"), + ], + metadata={"externalId": "grp-ext"}, + ) + final_team = LiteLLM_TeamTable( + team_id=group_id, + team_alias="Group", + members_with_roles=[Member(user_id="alice", role="user")], + metadata={"externalId": "grp-ext"}, + ) + + patch_ops = SCIMPatchOp( + schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + Operations=[SCIMPatchOperation(op="replace", path="members", value=[{"value": "alice"}])], + ) + + mock_prisma_client = mocker.MagicMock() + mock_prisma_client.db = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable = mocker.MagicMock() + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock( + side_effect=[snapshot_team, refreshed_team, final_team] + ) + mock_prisma_client.db.litellm_teamtable.update = AsyncMock(return_value=final_team) + mock_prisma_client.db.litellm_usertable = mocker.MagicMock() + mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=mocker.MagicMock()) + + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception", + AsyncMock(return_value=mock_prisma_client), + ) + patch_membership_mock = mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2.patch_team_membership", + AsyncMock(), + ) + mocker.patch( + "litellm.proxy.management_endpoints.scim.scim_v2._recompute_scim_member_roles", + AsyncMock(), + ) + mocker.patch.object( + ScimTransformations, + "transform_litellm_team_to_scim_group", + AsyncMock( + return_value=SCIMGroup( + schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"], + id=group_id, + displayName="Group", + ) + ), + ) + + await patch_group(group_id=group_id, patch_ops=patch_ops) + + calls = patch_membership_mock.call_args_list + + removed_user_ids = { + call.kwargs["user_id"] for call in calls if call.kwargs.get("teams_ids_to_remove_user_from") == [group_id] + } + assert removed_user_ids == {"bob"} + + added_user_ids = { + call.kwargs["user_id"] for call in calls if call.kwargs.get("teams_ids_to_add_user_to") == [group_id] + } + assert added_user_ids == set() diff --git a/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py index 50817b6a4c2..d5d61341c93 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py @@ -1693,6 +1693,78 @@ async def test_update_team_members_list_duplicate_prevention(): assert len(mock_team.members_with_roles) == 1 +@pytest.mark.asyncio +async def test_add_team_members_reconciles_against_freshly_locked_row(): + """ + Regression: _add_team_members_to_team must build the new members_with_roles + from the row it re-reads under a lock inside the write transaction, not from + the stale complete_team_data snapshot captured at the start of the request. + + Two concurrent /team/member_add calls for the same team read the same + snapshot; without the locked re-read the losing write rewrites the whole + JSON array from its stale copy and silently drops the member the other call + already committed. Here the snapshot holds only "zed", a concurrent writer + has already committed "alice" (returned by the locked SELECT), and this call + adds "bob". The write must contain all three. + """ + from litellm.proxy.management_endpoints.team_endpoints import ( + _add_team_members_to_team, + ) + + stale_snapshot = LiteLLM_TeamTable( + team_id="test-team-lock", + members_with_roles=[Member(user_id="zed", role="user")], + ) + + freshly_committed = [ + {"user_id": "zed", "user_email": None, "role": "user"}, + {"user_id": "alice", "user_email": None, "role": "user"}, + ] + + captured: dict = {} + + async def _capture_update(where, data): + captured["data"] = data + return LiteLLM_TeamTable( + team_id="test-team-lock", + members_with_roles=json.loads(data["members_with_roles"]), + ) + + tx = MagicMock() + tx.query_raw = AsyncMock(return_value=[{"members_with_roles": freshly_committed}]) + tx.litellm_teamtable.update = AsyncMock(side_effect=_capture_update) + + tx_cm = MagicMock() + tx_cm.__aenter__ = AsyncMock(return_value=tx) + tx_cm.__aexit__ = AsyncMock(return_value=None) + + prisma_client = MagicMock() + prisma_client.tx = MagicMock(return_value=tx_cm) + + with patch( + "litellm.proxy.management_endpoints.team_endpoints._process_team_members", + new=AsyncMock(return_value=([], [])), + ): + updated_team, _, _ = await _add_team_members_to_team( + data=TeamMemberAddRequest( + team_id="test-team-lock", + member=Member(user_id="bob", role="user"), + ), + complete_team_data=stale_snapshot, + prisma_client=cast(object, prisma_client), + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN), + litellm_proxy_admin_name="admin", + ) + + written_ids = sorted(m["user_id"] for m in json.loads(captured["data"]["members_with_roles"])) + assert written_ids == ["alice", "bob", "zed"] + + lock_reads = [call for call in tx.query_raw.call_args_list if "FOR UPDATE" in str(call.args[0])] + assert lock_reads, "expected a SELECT ... FOR UPDATE row-lock read before the write" + + assert [m.user_id for m in updated_team.members_with_roles] == ["zed", "alice", "bob"] + + def test_add_new_models_to_team_with_existing_models(): """ Test add_new_models_to_team function with existing models diff --git a/tests/test_litellm/repositories/test_repositories.py b/tests/test_litellm/repositories/test_repositories.py index af2eea823f4..6308faf8fc7 100644 --- a/tests/test_litellm/repositories/test_repositories.py +++ b/tests/test_litellm/repositories/test_repositories.py @@ -5,7 +5,7 @@ Tests for gateway repository layer. import json from datetime import datetime from typing import Any, Dict, List, Optional -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -499,6 +499,42 @@ class TestTeamRepository: assert team.team_id == "team-123" assert team.team_alias == "Engineering" + @pytest.mark.asyncio + @pytest.mark.parametrize( + "raw_value, expected_ids", + [ + ( + [ + {"user_id": "a", "role": "user"}, + {"user_id": "b", "role": "admin"}, + ], + ["a", "b"], + ), + (json.dumps([{"user_id": "a", "role": "user"}]), ["a"]), + ({}, []), + (None, []), + ], + ) + async def test_get_members_with_roles_locked(self, repo, raw_value, expected_ids): + tx = MagicMock() + tx.query_raw = AsyncMock(return_value=[{"members_with_roles": raw_value}]) + + members = await repo.get_members_with_roles_locked(tx, "team-1") + + assert [m.user_id for m in members] == expected_ids + sql = tx.query_raw.call_args.args[0] + assert "FOR UPDATE" in sql + assert tx.query_raw.call_args.args[1] == "team-1" + + @pytest.mark.asyncio + async def test_get_members_with_roles_locked_missing_row(self, repo): + tx = MagicMock() + tx.query_raw = AsyncMock(return_value=[]) + + members = await repo.get_members_with_roles_locked(tx, "missing") + + assert members == [] + @pytest.mark.asyncio async def test_create_team_all_fields(self, repo): team = await repo.create_team(