mirror of
https://github.com/tiennm99/litellm.git
synced 2026-08-24 00:28:32 +00:00
patching with feature flag
This commit is contained in:
yuneng-jiang
@@ -206,23 +206,52 @@ def _build_scim_metadata(
|
||||
return metadata
|
||||
|
||||
|
||||
async def _get_scim_upsert_user_setting() -> bool:
|
||||
"""
|
||||
Get the scim_upsert_user setting from litellm_settings.
|
||||
|
||||
Returns:
|
||||
True if scim_upsert_user is not set or is True (default behavior),
|
||||
False if scim_upsert_user is explicitly set to False (SCIM 2.0 strict mode)
|
||||
"""
|
||||
try:
|
||||
from litellm.proxy.proxy_server import proxy_config
|
||||
|
||||
config = await proxy_config.get_config()
|
||||
litellm_settings = config.get("litellm_settings", {}) or {}
|
||||
scim_upsert_user = litellm_settings.get("scim_upsert_user", True)
|
||||
|
||||
# Default to True if not set (backward compatibility)
|
||||
return bool(scim_upsert_user)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.warning(
|
||||
f"Error reading scim_upsert_user setting, defaulting to True: {e}"
|
||||
)
|
||||
# Default to True for backward compatibility
|
||||
return True
|
||||
|
||||
|
||||
async def _extract_group_member_ids(group: SCIMGroup) -> GroupMemberExtractionResult:
|
||||
"""
|
||||
Extract member IDs from SCIMGroup, validating that all users exist.
|
||||
|
||||
Per SCIM 2.0 protocol, groups should only reference existing users.
|
||||
Users must be created via POST /Users before being added to groups.
|
||||
Behavior depends on litellm_settings.scim_upsert_user:
|
||||
- If True (default): Creates users that don't exist (backward compatible)
|
||||
- If False: Rejects non-existent users per SCIM 2.0 protocol
|
||||
|
||||
Returns:
|
||||
GroupMemberExtractionResult with existing members and all member IDs
|
||||
GroupMemberExtractionResult with existing members, created users, and all member IDs
|
||||
|
||||
Raises:
|
||||
HTTPException: If any member user does not exist (400 Bad Request)
|
||||
HTTPException: If scim_upsert_user is False and any member user does not exist (400 Bad Request)
|
||||
"""
|
||||
prisma_client = await _get_prisma_client_or_raise_exception()
|
||||
existing_member_ids = []
|
||||
created_users = [] # Always empty - users must exist before group membership
|
||||
created_users = []
|
||||
all_member_ids = []
|
||||
|
||||
# Check the feature flag
|
||||
scim_upsert_user = await _get_scim_upsert_user_setting()
|
||||
|
||||
if group.members:
|
||||
for member in group.members:
|
||||
@@ -246,16 +275,26 @@ async def _extract_group_member_ids(group: SCIMGroup) -> GroupMemberExtractionRe
|
||||
existing_member_ids.append(user_id)
|
||||
all_member_ids.append(user_id)
|
||||
else:
|
||||
# User doesn't exist - reject per SCIM 2.0 protocol
|
||||
# This prevents security issues where users not assigned to app
|
||||
# get provisioned via group membership
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": f"User with ID '{user_id}' does not exist. "
|
||||
"Please create the user first via POST /Users before adding to group."
|
||||
},
|
||||
)
|
||||
if scim_upsert_user:
|
||||
# Create the user if they don't exist (backward compatible behavior)
|
||||
created_user = await _create_user_if_not_exists(
|
||||
user_id=user_id, created_via="scim_group_membership"
|
||||
)
|
||||
if created_user:
|
||||
created_users.append(created_user)
|
||||
all_member_ids.append(user_id)
|
||||
# If creation failed, user is skipped (logged in helper)
|
||||
else:
|
||||
# User doesn't exist - reject per SCIM 2.0 protocol
|
||||
# This prevents security issues where users not assigned to app
|
||||
# get provisioned via group membership
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": f"User with ID '{user_id}' does not exist. "
|
||||
"Please create the user first via POST /Users before adding to group."
|
||||
},
|
||||
)
|
||||
|
||||
return GroupMemberExtractionResult(
|
||||
existing_member_ids=existing_member_ids,
|
||||
@@ -328,7 +367,7 @@ async def _create_user_if_not_exists(
|
||||
|
||||
new_user_request = NewUserRequest(
|
||||
user_id=user_id,
|
||||
user_email=None, # We don't have email from group membership
|
||||
user_email=user_id, # We don't have email from group membership
|
||||
user_alias=None,
|
||||
teams=[], # Teams will be added separately
|
||||
metadata={"created_via": created_via},
|
||||
@@ -1220,7 +1259,9 @@ async def _process_group_patch_operations(
|
||||
elif path.startswith("members"):
|
||||
# Handle member operations
|
||||
member_values = _extract_group_values(value)
|
||||
# Validate all users exist - per SCIM 2.0, users must exist before group membership
|
||||
# Check the feature flag
|
||||
scim_upsert_user = await _get_scim_upsert_user_setting()
|
||||
# Validate all users exist or create them based on feature flag
|
||||
valid_members = []
|
||||
for member_id in member_values:
|
||||
# Validate member_id is not empty
|
||||
@@ -1238,14 +1279,23 @@ async def _process_group_patch_operations(
|
||||
if user:
|
||||
valid_members.append(member_id)
|
||||
else:
|
||||
# User doesn't exist - reject per SCIM 2.0 protocol
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": f"User with ID '{member_id}' does not exist. "
|
||||
"Please create the user first via POST /Users before adding to group."
|
||||
},
|
||||
)
|
||||
if scim_upsert_user:
|
||||
# Create the user if they don't exist (backward compatible behavior)
|
||||
created_user = await _create_user_if_not_exists(
|
||||
user_id=member_id, created_via="scim_group_patch"
|
||||
)
|
||||
if created_user:
|
||||
valid_members.append(member_id)
|
||||
# If creation failed, user is skipped (logged in helper)
|
||||
else:
|
||||
# User doesn't exist - reject per SCIM 2.0 protocol
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": f"User with ID '{member_id}' does not exist. "
|
||||
"Please create the user first via POST /Users before adding to group."
|
||||
},
|
||||
)
|
||||
|
||||
if op_type == "replace":
|
||||
final_members = set(valid_members)
|
||||
|
||||
@@ -3,10 +3,12 @@ from unittest.mock import AsyncMock
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from litellm.proxy._types import LitellmUserRoles, NewUserRequest, ProxyException
|
||||
from litellm.proxy._types import LitellmUserRoles, NewUserRequest, NewUserResponse, ProxyException
|
||||
from litellm.proxy.management_endpoints.scim.scim_v2 import (
|
||||
UserProvisionerHelpers,
|
||||
_extract_group_member_ids,
|
||||
_handle_team_membership_changes,
|
||||
_process_group_patch_operations,
|
||||
create_group,
|
||||
create_user,
|
||||
get_service_provider_config,
|
||||
@@ -913,12 +915,23 @@ async def test_update_group_e2e(mocker):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_group_with_nonexistent_users_rejects(mocker):
|
||||
async def test_create_group_with_nonexistent_users_rejects(mocker, monkeypatch):
|
||||
"""
|
||||
Test that creating a group with non-existent users is rejected.
|
||||
Test that creating a group with non-existent users is rejected when scim_upsert_user is False.
|
||||
Per SCIM 2.0 protocol, users must exist before being added to groups.
|
||||
This prevents security issues where users not assigned to app get provisioned via group membership.
|
||||
"""
|
||||
# Mock the feature flag to False (SCIM 2.0 strict mode)
|
||||
async def mock_get_config():
|
||||
return {
|
||||
"litellm_settings": {
|
||||
"scim_upsert_user": False
|
||||
}
|
||||
}
|
||||
|
||||
from litellm.proxy.proxy_server import proxy_config
|
||||
monkeypatch.setattr(proxy_config, "get_config", mock_get_config)
|
||||
|
||||
# Test data
|
||||
group_id = "test-group-123"
|
||||
scim_group = SCIMGroup(
|
||||
@@ -962,22 +975,33 @@ async def test_create_group_with_nonexistent_users_rejects(mocker):
|
||||
AsyncMock(return_value=mock_prisma_client)
|
||||
)
|
||||
|
||||
# Execute the create_group function - should raise HTTPException
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
# Execute the create_group function - should raise ProxyException
|
||||
with pytest.raises(ProxyException) as exc_info:
|
||||
await create_group(group=scim_group)
|
||||
|
||||
# Verify it's a 400 Bad Request
|
||||
assert exc_info.value.status_code == 400
|
||||
assert "does not exist" in str(exc_info.value.detail)
|
||||
assert "new-user-1" in str(exc_info.value.detail) or "new-user-2" in str(exc_info.value.detail)
|
||||
assert int(exc_info.value.code) == 400
|
||||
assert "does not exist" in str(exc_info.value.message)
|
||||
assert "new-user-1" in str(exc_info.value.message) or "new-user-2" in str(exc_info.value.message)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_group_with_nonexistent_users_rejects(mocker):
|
||||
async def test_update_group_with_nonexistent_users_rejects(mocker, monkeypatch):
|
||||
"""
|
||||
Test that updating a group with non-existent users is rejected.
|
||||
Test that updating a group with non-existent users is rejected when scim_upsert_user is False.
|
||||
Per SCIM 2.0 protocol, users must exist before being added to groups.
|
||||
"""
|
||||
# Mock the feature flag to False (SCIM 2.0 strict mode)
|
||||
async def mock_get_config():
|
||||
return {
|
||||
"litellm_settings": {
|
||||
"scim_upsert_user": False
|
||||
}
|
||||
}
|
||||
|
||||
from litellm.proxy.proxy_server import proxy_config
|
||||
monkeypatch.setattr(proxy_config, "get_config", mock_get_config)
|
||||
|
||||
# Test data
|
||||
group_id = "existing-group-456"
|
||||
|
||||
@@ -1039,11 +1063,358 @@ async def test_update_group_with_nonexistent_users_rejects(mocker):
|
||||
AsyncMock(return_value=mock_existing_team)
|
||||
)
|
||||
|
||||
# Execute the update_group function - should raise HTTPException
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
# Execute the update_group function - should raise ProxyException
|
||||
with pytest.raises(ProxyException) as exc_info:
|
||||
await update_group(group_id=group_id, group=scim_group_update)
|
||||
|
||||
# Verify it's a 400 Bad Request
|
||||
assert int(exc_info.value.code) == 400
|
||||
assert "does not exist" in str(exc_info.value.message)
|
||||
assert "new-user-3" in str(exc_info.value.message) or "new-user-4" in str(exc_info.value.message)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_group_with_nonexistent_users_creates_when_flag_true(mocker, monkeypatch):
|
||||
"""
|
||||
Test that creating a group with non-existent users creates them when scim_upsert_user is True.
|
||||
This preserves backward compatible behavior.
|
||||
"""
|
||||
# Mock the feature flag to True (backward compatible mode)
|
||||
async def mock_get_config():
|
||||
return {
|
||||
"litellm_settings": {
|
||||
"scim_upsert_user": True
|
||||
}
|
||||
}
|
||||
|
||||
from litellm.proxy.proxy_server import proxy_config
|
||||
monkeypatch.setattr(proxy_config, "get_config", mock_get_config)
|
||||
|
||||
# Test data
|
||||
group_id = "test-group-123"
|
||||
scim_group = SCIMGroup(
|
||||
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
|
||||
id=group_id,
|
||||
displayName="Test Group",
|
||||
members=[
|
||||
SCIMMember(value="existing-user", display="Existing User"), # This user exists
|
||||
SCIMMember(value="new-user-1", display="New User 1"), # This user doesn't exist - should be created
|
||||
SCIMMember(value="new-user-2", display="New User 2"), # This user doesn't exist - should be created
|
||||
]
|
||||
)
|
||||
|
||||
# Mock prisma client
|
||||
mock_prisma_client = mocker.MagicMock()
|
||||
mock_prisma_client.db = mocker.MagicMock()
|
||||
mock_prisma_client.db.litellm_teamtable = mocker.MagicMock()
|
||||
mock_prisma_client.db.litellm_usertable = mocker.MagicMock()
|
||||
|
||||
# Mock team operations - team doesn't exist yet
|
||||
mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=None)
|
||||
|
||||
# Mock user lookup - only existing-user exists initially
|
||||
def mock_user_lookup(where):
|
||||
user_id = where["user_id"]
|
||||
if user_id == "existing-user":
|
||||
mock_user = mocker.MagicMock()
|
||||
mock_user.user_id = user_id
|
||||
return mock_user
|
||||
return None # new-user-1 and new-user-2 don't exist
|
||||
|
||||
mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(side_effect=mock_user_lookup)
|
||||
|
||||
# Mock user creation
|
||||
created_user_1 = NewUserResponse(user_id="new-user-1", key="test-key-1")
|
||||
created_user_2 = NewUserResponse(user_id="new-user-2", key="test-key-2")
|
||||
mock_create_user = mocker.patch(
|
||||
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
|
||||
AsyncMock(side_effect=[created_user_1, created_user_2])
|
||||
)
|
||||
|
||||
# Mock new_team
|
||||
mock_team = mocker.MagicMock()
|
||||
mock_team.team_id = group_id
|
||||
mock_new_team = mocker.patch(
|
||||
"litellm.proxy.management_endpoints.scim.scim_v2.new_team",
|
||||
AsyncMock(return_value=mock_team)
|
||||
)
|
||||
|
||||
# Mock transformation
|
||||
mock_scim_group = SCIMGroup(
|
||||
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
|
||||
id=group_id,
|
||||
displayName="Test Group",
|
||||
members=[]
|
||||
)
|
||||
mocker.patch(
|
||||
"litellm.proxy.management_endpoints.scim.scim_v2.ScimTransformations.transform_litellm_team_to_scim_group",
|
||||
AsyncMock(return_value=mock_scim_group)
|
||||
)
|
||||
|
||||
# Mock dependencies
|
||||
mocker.patch(
|
||||
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
|
||||
AsyncMock(return_value=mock_prisma_client)
|
||||
)
|
||||
|
||||
# Execute the create_group function - should succeed
|
||||
result = await create_group(group=scim_group)
|
||||
|
||||
# Verify users were created
|
||||
assert mock_create_user.call_count == 2
|
||||
assert mock_create_user.call_args_list[0].kwargs['user_id'] == "new-user-1"
|
||||
assert mock_create_user.call_args_list[1].kwargs['user_id'] == "new-user-2"
|
||||
|
||||
# Verify team was created
|
||||
mock_new_team.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_extract_group_member_ids_with_flag_true_creates_users(mocker, monkeypatch):
|
||||
"""
|
||||
Test that _extract_group_member_ids creates users when scim_upsert_user is True.
|
||||
"""
|
||||
# Mock the feature flag to True (backward compatible mode)
|
||||
async def mock_get_config():
|
||||
return {
|
||||
"litellm_settings": {
|
||||
"scim_upsert_user": True
|
||||
}
|
||||
}
|
||||
|
||||
from litellm.proxy.proxy_server import proxy_config
|
||||
monkeypatch.setattr(proxy_config, "get_config", mock_get_config)
|
||||
|
||||
# Test data
|
||||
scim_group = SCIMGroup(
|
||||
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
|
||||
id="test-group",
|
||||
displayName="Test Group",
|
||||
members=[
|
||||
SCIMMember(value="existing-user", display="Existing User"), # This user exists
|
||||
SCIMMember(value="new-user-1", display="New User 1"), # This user doesn't exist - should be created
|
||||
]
|
||||
)
|
||||
|
||||
# Mock prisma client
|
||||
mock_prisma_client = mocker.MagicMock()
|
||||
mock_prisma_client.db = mocker.MagicMock()
|
||||
mock_prisma_client.db.litellm_usertable = mocker.MagicMock()
|
||||
|
||||
# Mock user lookup - only existing-user exists initially
|
||||
def mock_user_lookup(where):
|
||||
user_id = where["user_id"]
|
||||
if user_id == "existing-user":
|
||||
mock_user = mocker.MagicMock()
|
||||
mock_user.user_id = user_id
|
||||
return mock_user
|
||||
return None # new-user-1 doesn't exist
|
||||
|
||||
mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(side_effect=mock_user_lookup)
|
||||
|
||||
# Mock user creation
|
||||
created_user = NewUserResponse(user_id="new-user-1", key="test-key-1")
|
||||
mock_create_user = mocker.patch(
|
||||
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
|
||||
AsyncMock(return_value=created_user)
|
||||
)
|
||||
|
||||
# Mock dependencies
|
||||
mocker.patch(
|
||||
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
|
||||
AsyncMock(return_value=mock_prisma_client)
|
||||
)
|
||||
|
||||
# Execute the function
|
||||
result = await _extract_group_member_ids(scim_group)
|
||||
|
||||
# Verify result
|
||||
assert "existing-user" in result.existing_member_ids
|
||||
assert "existing-user" in result.all_member_ids
|
||||
assert "new-user-1" in result.all_member_ids
|
||||
assert len(result.created_users) == 1
|
||||
|
||||
# Verify user was created
|
||||
mock_create_user.assert_called_once_with(
|
||||
user_id="new-user-1",
|
||||
created_via="scim_group_membership"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_extract_group_member_ids_with_flag_false_rejects(mocker, monkeypatch):
|
||||
"""
|
||||
Test that _extract_group_member_ids rejects non-existent users when scim_upsert_user is False.
|
||||
"""
|
||||
# Mock the feature flag to False (SCIM 2.0 strict mode)
|
||||
async def mock_get_config():
|
||||
return {
|
||||
"litellm_settings": {
|
||||
"scim_upsert_user": False
|
||||
}
|
||||
}
|
||||
|
||||
from litellm.proxy.proxy_server import proxy_config
|
||||
monkeypatch.setattr(proxy_config, "get_config", mock_get_config)
|
||||
|
||||
# Test data
|
||||
scim_group = SCIMGroup(
|
||||
schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
|
||||
id="test-group",
|
||||
displayName="Test Group",
|
||||
members=[
|
||||
SCIMMember(value="existing-user", display="Existing User"), # This user exists
|
||||
SCIMMember(value="new-user-1", display="New User 1"), # This user doesn't exist - should be rejected
|
||||
]
|
||||
)
|
||||
|
||||
# Mock prisma client
|
||||
mock_prisma_client = mocker.MagicMock()
|
||||
mock_prisma_client.db = mocker.MagicMock()
|
||||
mock_prisma_client.db.litellm_usertable = mocker.MagicMock()
|
||||
|
||||
# Mock user lookup - only existing-user exists
|
||||
def mock_user_lookup(where):
|
||||
user_id = where["user_id"]
|
||||
if user_id == "existing-user":
|
||||
mock_user = mocker.MagicMock()
|
||||
mock_user.user_id = user_id
|
||||
return mock_user
|
||||
return None # new-user-1 doesn't exist
|
||||
|
||||
mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(side_effect=mock_user_lookup)
|
||||
|
||||
# Mock dependencies
|
||||
mocker.patch(
|
||||
"litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
|
||||
AsyncMock(return_value=mock_prisma_client)
|
||||
)
|
||||
|
||||
# Execute the function - should raise HTTPException
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await _extract_group_member_ids(scim_group)
|
||||
|
||||
# Verify it's a 400 Bad Request
|
||||
assert exc_info.value.status_code == 400
|
||||
assert "does not exist" in str(exc_info.value.detail)
|
||||
assert "new-user-3" in str(exc_info.value.detail) or "new-user-4" in str(exc_info.value.detail)
|
||||
assert "new-user-1" in str(exc_info.value.detail)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_group_patch_operations_with_flag_true_creates_users(mocker, monkeypatch):
|
||||
"""
|
||||
Test that _process_group_patch_operations creates users when scim_upsert_user is True.
|
||||
"""
|
||||
# Mock the feature flag to True (backward compatible mode)
|
||||
async def mock_get_config():
|
||||
return {
|
||||
"litellm_settings": {
|
||||
"scim_upsert_user": True
|
||||
}
|
||||
}
|
||||
|
||||
from litellm.proxy.proxy_server import proxy_config
|
||||
monkeypatch.setattr(proxy_config, "get_config", mock_get_config)
|
||||
|
||||
# Test data
|
||||
patch_ops = SCIMPatchOp(
|
||||
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="add",
|
||||
path="members",
|
||||
value=[{"value": "new-user-1"}]
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
# Mock existing team
|
||||
mock_existing_team = mocker.MagicMock()
|
||||
mock_existing_team.members = []
|
||||
mock_existing_team.metadata = {}
|
||||
|
||||
# Mock prisma client
|
||||
mock_prisma_client = mocker.MagicMock()
|
||||
mock_prisma_client.db = mocker.MagicMock()
|
||||
mock_prisma_client.db.litellm_usertable = mocker.MagicMock()
|
||||
|
||||
# Mock user lookup - new-user-1 doesn't exist
|
||||
mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=None)
|
||||
|
||||
# Mock user creation
|
||||
created_user = NewUserResponse(user_id="new-user-1", key="test-key-1")
|
||||
mock_create_user = mocker.patch(
|
||||
"litellm.proxy.management_endpoints.scim.scim_v2._create_user_if_not_exists",
|
||||
AsyncMock(return_value=created_user)
|
||||
)
|
||||
|
||||
# Execute the function
|
||||
update_data, final_members = await _process_group_patch_operations(
|
||||
patch_ops=patch_ops,
|
||||
existing_team=mock_existing_team,
|
||||
prisma_client=mock_prisma_client
|
||||
)
|
||||
|
||||
# Verify result
|
||||
assert "new-user-1" in final_members
|
||||
|
||||
# Verify user was created
|
||||
mock_create_user.assert_called_once_with(
|
||||
user_id="new-user-1",
|
||||
created_via="scim_group_patch"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_group_patch_operations_with_flag_false_rejects(mocker, monkeypatch):
|
||||
"""
|
||||
Test that _process_group_patch_operations rejects non-existent users when scim_upsert_user is False.
|
||||
"""
|
||||
# Mock the feature flag to False (SCIM 2.0 strict mode)
|
||||
async def mock_get_config():
|
||||
return {
|
||||
"litellm_settings": {
|
||||
"scim_upsert_user": False
|
||||
}
|
||||
}
|
||||
|
||||
from litellm.proxy.proxy_server import proxy_config
|
||||
monkeypatch.setattr(proxy_config, "get_config", mock_get_config)
|
||||
|
||||
# Test data
|
||||
patch_ops = SCIMPatchOp(
|
||||
schemas=["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="add",
|
||||
path="members",
|
||||
value=[{"value": "new-user-1"}]
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
# Mock existing team
|
||||
mock_existing_team = mocker.MagicMock()
|
||||
mock_existing_team.members = []
|
||||
mock_existing_team.metadata = {}
|
||||
|
||||
# Mock prisma client
|
||||
mock_prisma_client = mocker.MagicMock()
|
||||
mock_prisma_client.db = mocker.MagicMock()
|
||||
mock_prisma_client.db.litellm_usertable = mocker.MagicMock()
|
||||
|
||||
# Mock user lookup - new-user-1 doesn't exist
|
||||
mock_prisma_client.db.litellm_usertable.find_unique = AsyncMock(return_value=None)
|
||||
|
||||
# Execute the function - should raise HTTPException
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await _process_group_patch_operations(
|
||||
patch_ops=patch_ops,
|
||||
existing_team=mock_existing_team,
|
||||
prisma_client=mock_prisma_client
|
||||
)
|
||||
|
||||
# Verify it's a 400 Bad Request
|
||||
assert exc_info.value.status_code == 400
|
||||
assert "does not exist" in str(exc_info.value.detail)
|
||||
assert "new-user-1" in str(exc_info.value.detail)
|
||||
|
||||
Reference in New Issue
Block a user