scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-14 23:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

fix(team): make team member add atomic to prevent concurrent-add member loss (#34185)

Browse source 
_add_team_members_to_team reconciled membership by reading the complete_team_data
snapshot captured at the start of team_member_add, appending in memory, and
writing the whole members_with_roles array back. Two concurrent /team/member_add
calls for the same team read the same snapshot, so the last write wins and one
member is silently lost. This affects every concurrent team member add, including
the SCIM group PATCH op:add path that routes through team_member_add

Reconcile members_with_roles inside a transaction that locks the team row with
SELECT ... FOR UPDATE before re-reading the current membership, so concurrent
writers serialize on the row lock and each appends onto the other's committed
result. The interactive transaction is exposed through a thin PrismaClient.tx()
passthrough and the locked read is encapsulated in
TeamRepository.get_members_with_roles_locked, and the SCIM group PATCH applies
membership as deltas so concurrent adds are not clobbered
This commit is contained in:
Yassin Kortam 2026-07-22 14:47:22 -07:00 • committed by GitHub
parent 3f98f62748
commit c6b2f111a6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 448 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
litellm/proxy/management_endpoints/scim/scim_v2.py
View file

@ -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

37
litellm/proxy/management_endpoints/team_endpoints.py
View file

@ -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

9
litellm/proxy/utils.py
View file

@ -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.

29
litellm/repositories/team_repository.py
View file

@ -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)

83
tests/proxy_unit_tests/test_proxy_server.py
View file

@ -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

197
tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py
View file

@ -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()

72
tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py
View file

@ -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

38
tests/test_litellm/repositories/test_repositories.py
View file

@ -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(