mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-25 05:54:53 +00:00
fix(scim)!: store PATCH enterprise extension attributes in metadata.scim_enterprise (#46738)
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Yassin Kortam
GitHub
parent
9b4f44135d
commit
48763ccb50
2 changed files with 805 additions and 10 deletions
|
|
@ -5,13 +5,13 @@ This is an enterprise feature and requires a premium license.
|
|||
"""
|
||||
|
||||
import re
|
||||
from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
|
||||
from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping, Sequence
|
||||
from copy import deepcopy
|
||||
from dataclasses import dataclass
|
||||
from functools import partial
|
||||
from functools import partial, reduce
|
||||
from itertools import chain, groupby
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Final, NamedTuple, Protocol, TypeAlias, overload
|
||||
from typing import TYPE_CHECKING, Final, NamedTuple, NoReturn, Protocol, TypeAlias, overload
|
||||
|
||||
from fastapi import (
|
||||
APIRouter,
|
||||
|
|
@ -24,6 +24,7 @@ from fastapi import (
|
|||
Response,
|
||||
)
|
||||
from pydantic import TypeAdapter, ValidationError
|
||||
from pydantic.fields import FieldInfo
|
||||
from typing_extensions import ReadOnly, TypedDict, assert_never
|
||||
|
||||
import litellm
|
||||
|
|
@ -2271,6 +2272,217 @@ def _handle_multi_valued_attribute_update(path: str, op_type: str, value: object
|
|||
metadata[metadata_key] = dumped
|
||||
|
||||
|
||||
def _enterprise_extension_sub_path(path: str) -> str | None:
|
||||
schema: Final = SCIM_ENTERPRISE_USER_SCHEMA
|
||||
if path.casefold() == schema.casefold():
|
||||
return ""
|
||||
prefix: Final = f"{schema}:"
|
||||
if path.casefold().startswith(prefix.casefold()):
|
||||
return path[len(prefix) :]
|
||||
return None
|
||||
|
||||
|
||||
def _scim_model_field_names(model_fields: Mapping[str, FieldInfo]) -> Mapping[str, str]:
|
||||
return {
|
||||
**{field_name.casefold(): field_name for field_name in model_fields},
|
||||
**{
|
||||
field.alias.casefold(): field_name
|
||||
for field_name, field in model_fields.items()
|
||||
if isinstance(field.alias, str)
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _canonicalize_scim_model_data(
|
||||
value: Mapping[str, object],
|
||||
model_fields: Mapping[str, FieldInfo],
|
||||
) -> Mapping[str, object]:
|
||||
field_names: Final = _scim_model_field_names(model_fields)
|
||||
return {field_names.get(key.casefold(), key): field_value for key, field_value in value.items()}
|
||||
|
||||
|
||||
def _canonicalize_enterprise_data(value: Mapping[str, object]) -> Mapping[str, object]:
|
||||
enterprise_data: Final = _canonicalize_scim_model_data(value, SCIMEnterpriseUser.model_fields)
|
||||
manager: Final = enterprise_data.get("manager")
|
||||
if isinstance(manager, str):
|
||||
return {**enterprise_data, "manager": {"value": manager}}
|
||||
if not isinstance(manager, dict):
|
||||
return enterprise_data
|
||||
return {
|
||||
**enterprise_data,
|
||||
"manager": _canonicalize_scim_model_data(
|
||||
_JSON_OBJECT_ADAPTER.validate_python(manager),
|
||||
SCIMUserManager.model_fields,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _InvalidEnterpriseValue:
|
||||
detail: str
|
||||
|
||||
|
||||
def _merge_enterprise_data(
|
||||
existing: Mapping[str, object],
|
||||
incoming: Mapping[str, object],
|
||||
incoming_manager_is_object: bool,
|
||||
) -> Mapping[str, object]:
|
||||
merged: Final[Mapping[str, object]] = {**existing, **incoming}
|
||||
existing_manager: Final = existing.get("manager")
|
||||
incoming_manager: Final = incoming.get("manager")
|
||||
if incoming_manager_is_object and isinstance(existing_manager, dict) and isinstance(incoming_manager, dict):
|
||||
existing_manager_data: Final = _canonicalize_scim_model_data(
|
||||
_JSON_OBJECT_ADAPTER.validate_python(existing_manager),
|
||||
SCIMUserManager.model_fields,
|
||||
)
|
||||
incoming_manager_data: Final = _JSON_OBJECT_ADAPTER.validate_python(incoming_manager)
|
||||
return {**merged, "manager": {**existing_manager_data, **incoming_manager_data}}
|
||||
return merged
|
||||
|
||||
|
||||
def _validate_enterprise_update(
|
||||
value: Mapping[str, object],
|
||||
) -> Mapping[str, object] | None | _InvalidEnterpriseValue:
|
||||
try:
|
||||
enterprise: Final = SCIMEnterpriseUser.model_validate(value)
|
||||
except ValidationError:
|
||||
return _InvalidEnterpriseValue(detail="Invalid value for the enterprise extension")
|
||||
dumped: Final = enterprise.model_dump(by_alias=True, exclude_none=True)
|
||||
return dumped if dumped else None
|
||||
|
||||
|
||||
def _raise_invalid_enterprise_value(error: _InvalidEnterpriseValue) -> NoReturn:
|
||||
raise HTTPException(status_code=400, detail={"error": error.detail})
|
||||
|
||||
|
||||
def _handle_enterprise_update(
|
||||
sub_path: str,
|
||||
op_type: str,
|
||||
value: object,
|
||||
existing_enterprise: Mapping[str, object],
|
||||
) -> Mapping[str, object] | None | _InvalidEnterpriseValue:
|
||||
if op_type == "remove" and not sub_path:
|
||||
return None
|
||||
|
||||
if not sub_path:
|
||||
if not isinstance(value, dict):
|
||||
return _InvalidEnterpriseValue(detail="The enterprise extension value must be an object")
|
||||
incoming_value: Final = _JSON_OBJECT_ADAPTER.validate_python(value)
|
||||
incoming: Final = _canonicalize_enterprise_data(incoming_value)
|
||||
incoming_manager_is_object: Final = isinstance(
|
||||
_canonicalize_scim_model_data(incoming_value, SCIMEnterpriseUser.model_fields).get("manager"),
|
||||
dict,
|
||||
)
|
||||
return _validate_enterprise_update(
|
||||
_merge_enterprise_data(existing_enterprise, incoming, incoming_manager_is_object)
|
||||
)
|
||||
|
||||
path_parts: Final = sub_path.split(".")
|
||||
if len(path_parts) == 1:
|
||||
field_name: Final = _scim_model_field_names(SCIMEnterpriseUser.model_fields).get(sub_path.casefold())
|
||||
if field_name is None:
|
||||
return _validate_enterprise_update(existing_enterprise)
|
||||
if op_type == "remove":
|
||||
return _validate_enterprise_update(
|
||||
{key: attr for key, attr in existing_enterprise.items() if key != field_name}
|
||||
)
|
||||
if value is None:
|
||||
return _InvalidEnterpriseValue(detail=f"The {op_type} operation requires a 'value' member")
|
||||
field_incoming: Final = _canonicalize_enterprise_data({field_name: value})
|
||||
field_incoming_manager_is_object: Final = field_name == "manager" and isinstance(value, dict)
|
||||
return _validate_enterprise_update(
|
||||
_merge_enterprise_data(
|
||||
existing_enterprise,
|
||||
field_incoming,
|
||||
field_incoming_manager_is_object,
|
||||
)
|
||||
)
|
||||
|
||||
if len(path_parts) == 2 and path_parts[0].casefold() == "manager":
|
||||
manager_field: Final = _scim_model_field_names(SCIMUserManager.model_fields).get(path_parts[1].casefold())
|
||||
if manager_field is None:
|
||||
return _validate_enterprise_update(existing_enterprise)
|
||||
existing_manager: Final = existing_enterprise.get("manager")
|
||||
manager_data: Final[Mapping[str, object]] = (
|
||||
_canonicalize_scim_model_data(
|
||||
_JSON_OBJECT_ADAPTER.validate_python(existing_manager),
|
||||
SCIMUserManager.model_fields,
|
||||
)
|
||||
if isinstance(existing_manager, dict)
|
||||
else {}
|
||||
)
|
||||
if op_type == "remove":
|
||||
removed_manager: Final[Mapping[str, object]] = {
|
||||
key: attr for key, attr in manager_data.items() if key != manager_field
|
||||
}
|
||||
updated_enterprise: Final[Mapping[str, object]] = (
|
||||
{**existing_enterprise, "manager": removed_manager}
|
||||
if removed_manager
|
||||
else {key: attr for key, attr in existing_enterprise.items() if key != "manager"}
|
||||
)
|
||||
return _validate_enterprise_update(updated_enterprise)
|
||||
if value is None:
|
||||
return _InvalidEnterpriseValue(detail=f"The {op_type} operation requires a 'value' member")
|
||||
updated_manager: Final[Mapping[str, object]] = {**manager_data, manager_field: value}
|
||||
return _validate_enterprise_update({**existing_enterprise, "manager": updated_manager})
|
||||
|
||||
return _validate_enterprise_update(existing_enterprise)
|
||||
|
||||
|
||||
def _enterprise_update_for_pathless_key(
|
||||
key_and_value: tuple[str, object],
|
||||
*,
|
||||
op_type: str,
|
||||
) -> Iterator[tuple[str, str, object]]:
|
||||
key, value = key_and_value
|
||||
sub_path: Final = _enterprise_extension_sub_path(key)
|
||||
if sub_path is not None:
|
||||
yield sub_path, op_type, value
|
||||
|
||||
|
||||
def _enterprise_updates_for_operation(
|
||||
op: SCIMPatchOperation,
|
||||
) -> Iterator[tuple[str, str, object]]:
|
||||
path: Final = op.path.lower() if isinstance(op.path, str) else ""
|
||||
value: Final[object] = TypeAdapter(object).validate_python(op.value)
|
||||
if not path and isinstance(value, dict):
|
||||
pathless_values: Final[Mapping[str, object]] = _JSON_OBJECT_ADAPTER.validate_python(value)
|
||||
yield from chain.from_iterable(
|
||||
map(partial(_enterprise_update_for_pathless_key, op_type=op.op), pathless_values.items())
|
||||
)
|
||||
return
|
||||
sub_path: Final = _enterprise_extension_sub_path(path)
|
||||
if sub_path is not None:
|
||||
yield sub_path, op.op, value
|
||||
|
||||
|
||||
def _enterprise_updates(operations: Sequence[SCIMPatchOperation]) -> Iterator[tuple[str, str, object]]:
|
||||
yield from chain.from_iterable(map(_enterprise_updates_for_operation, operations))
|
||||
|
||||
|
||||
def _apply_enterprise_update(
|
||||
state: Mapping[str, object] | None | _InvalidEnterpriseValue,
|
||||
update: tuple[str, str, object],
|
||||
) -> Mapping[str, object] | None | _InvalidEnterpriseValue:
|
||||
if isinstance(state, _InvalidEnterpriseValue):
|
||||
return state
|
||||
sub_path, op_type, value = update
|
||||
existing_enterprise: Final[Mapping[str, object]] = {} if state is None else state
|
||||
return _handle_enterprise_update(sub_path, op_type, value, existing_enterprise)
|
||||
|
||||
|
||||
def _apply_enterprise_updates(
|
||||
updates: Sequence[tuple[str, str, object]],
|
||||
stored_enterprise: object,
|
||||
) -> Mapping[str, object] | None | _InvalidEnterpriseValue:
|
||||
initial_enterprise: Final[Mapping[str, object] | None] = (
|
||||
_canonicalize_enterprise_data(_JSON_OBJECT_ADAPTER.validate_python(stored_enterprise))
|
||||
if isinstance(stored_enterprise, dict)
|
||||
else None
|
||||
)
|
||||
return reduce(_apply_enterprise_update, updates, initial_enterprise)
|
||||
|
||||
|
||||
def _handle_generic_metadata(path: str, op_type: str, value: object, metadata: dict[str, object]) -> None:
|
||||
"""Handle generic metadata operations for unknown paths."""
|
||||
if op_type == "remove":
|
||||
|
|
@ -2285,20 +2497,37 @@ def _apply_patch_ops(
|
|||
) -> tuple[dict[str, object], set[str]]:
|
||||
"""Apply patch operations and return update data and final team set."""
|
||||
update_data: Final[dict[str, object]] = {}
|
||||
metadata: Final = existing_user.metadata or {}
|
||||
scim_metadata: Final = metadata.get("scim_metadata", {})
|
||||
stored_metadata: Final[object] = TypeAdapter(object).validate_python(existing_user.model_dump().get("metadata"))
|
||||
metadata: Final = _JSON_OBJECT_ADAPTER.validate_python(stored_metadata) if isinstance(stored_metadata, dict) else {}
|
||||
stored_scim_metadata: Final = metadata.get("scim_metadata")
|
||||
scim_metadata: Final = (
|
||||
_JSON_OBJECT_ADAPTER.validate_python(stored_scim_metadata) if isinstance(stored_scim_metadata, dict) else {}
|
||||
)
|
||||
enterprise_updates: Final = tuple(_enterprise_updates(patch_ops.Operations))
|
||||
enterprise: Final[Mapping[str, object] | None | _InvalidEnterpriseValue] = (
|
||||
_apply_enterprise_updates(
|
||||
enterprise_updates,
|
||||
metadata.get(SCIM_ENTERPRISE_METADATA_KEY),
|
||||
)
|
||||
if enterprise_updates
|
||||
else None
|
||||
)
|
||||
if isinstance(enterprise, _InvalidEnterpriseValue):
|
||||
_raise_invalid_enterprise_value(enterprise)
|
||||
|
||||
teams_set: Final[set[str]] = set(existing_user.teams or [])
|
||||
replace_team_set: set[str] | None = None
|
||||
|
||||
for op in patch_ops.Operations:
|
||||
path = (op.path or "").lower()
|
||||
path = op.path.lower() if isinstance(op.path, str) else ""
|
||||
value = op.value
|
||||
op_type = op.op
|
||||
|
||||
# Handle SCIM operations without path where value contains the fields
|
||||
if not path and isinstance(value, dict):
|
||||
for key, val in value.items():
|
||||
for key, val in _JSON_OBJECT_ADAPTER.validate_python(value).items():
|
||||
if _enterprise_extension_sub_path(key) is not None:
|
||||
continue
|
||||
key_lower = key.lower()
|
||||
if key_lower == "active":
|
||||
_handle_active_update(op_type, val, metadata)
|
||||
|
|
@ -2309,7 +2538,7 @@ def _apply_patch_ops(
|
|||
elif key_lower in SCIM_MULTI_VALUED_ATTRIBUTE_METADATA_KEYS:
|
||||
_handle_multi_valued_attribute_update(key_lower, op_type, val, metadata)
|
||||
elif key_lower == "name" and isinstance(val, dict):
|
||||
for name_key, name_val in val.items():
|
||||
for name_key, name_val in _JSON_OBJECT_ADAPTER.validate_python(val).items():
|
||||
name_key_lower = name_key.lower()
|
||||
if name_key_lower in ("givenname", "familyname"):
|
||||
_handle_name_update(
|
||||
|
|
@ -2334,12 +2563,22 @@ def _apply_patch_ops(
|
|||
new_replace_set = _handle_group_operations(op_type, value, teams_set, op.path)
|
||||
if new_replace_set is not None:
|
||||
replace_team_set = new_replace_set
|
||||
elif _enterprise_extension_sub_path(path) is not None:
|
||||
continue
|
||||
else:
|
||||
_handle_generic_metadata(path, op_type, value, metadata)
|
||||
|
||||
final_team_set: Final = replace_team_set if replace_team_set is not None else teams_set
|
||||
metadata["scim_metadata"] = scim_metadata
|
||||
update_data["metadata"] = metadata
|
||||
final_metadata: Final = (
|
||||
{
|
||||
**{key: metadata_value for key, metadata_value in metadata.items() if key != SCIM_ENTERPRISE_METADATA_KEY},
|
||||
"scim_metadata": scim_metadata,
|
||||
**({SCIM_ENTERPRISE_METADATA_KEY: enterprise} if enterprise is not None else {}),
|
||||
}
|
||||
if enterprise_updates
|
||||
else {**metadata, "scim_metadata": scim_metadata}
|
||||
)
|
||||
update_data["metadata"] = final_metadata
|
||||
return update_data, final_team_set
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,11 +1,17 @@
|
|||
from collections.abc import Mapping
|
||||
from typing import Final
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from pydantic import TypeAdapter
|
||||
|
||||
from litellm.proxy._types import LiteLLM_UserTable
|
||||
from litellm.proxy.management_endpoints.scim.scim_v2 import _apply_patch_ops, patch_user
|
||||
from litellm.types.proxy.management_endpoints.scim_v2 import (
|
||||
SCIM_ENTERPRISE_METADATA_KEY,
|
||||
SCIM_ENTERPRISE_USER_SCHEMA,
|
||||
SCIMEnterpriseUser,
|
||||
SCIMPatchOp,
|
||||
SCIMPatchOperation,
|
||||
SCIMUser,
|
||||
|
|
@ -536,3 +542,553 @@ def test_apply_patch_ops_replace_groups_empty_value_does_not_use_path_filter():
|
|||
_, final_team_set = _apply_patch_ops(existing_user=user, patch_ops=patch_ops)
|
||||
|
||||
assert final_team_set == set()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_patch_user_pathless_enterprise_extension_merges_existing_metadata():
|
||||
mock_user: Final = _user_with_metadata({SCIM_ENTERPRISE_METADATA_KEY: {"employeeNumber": "42"}})
|
||||
mock_client: Final = MagicMock()
|
||||
mock_db: Final = MagicMock()
|
||||
mock_client.db = mock_db
|
||||
mock_db.litellm_usertable.find_unique = AsyncMock(return_value=mock_user)
|
||||
|
||||
async def mock_update(*, where: Mapping[str, object], data: Mapping[str, object]) -> LiteLLM_UserTable:
|
||||
assert where == {"user_id": "user-mva"}
|
||||
metadata_json: Final = data["metadata"]
|
||||
assert isinstance(metadata_json, str)
|
||||
metadata: Final = TypeAdapter(dict[str, object]).validate_json(metadata_json)
|
||||
return _user_with_metadata(metadata)
|
||||
|
||||
mock_db.litellm_usertable.update = AsyncMock(side_effect=mock_update)
|
||||
mock_db.litellm_teamtable.find_unique = AsyncMock(return_value=None)
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="replace",
|
||||
value={
|
||||
SCIM_ENTERPRISE_USER_SCHEMA: {
|
||||
"department": "Eng",
|
||||
"costCenter": "CC-10",
|
||||
}
|
||||
},
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
with (
|
||||
patch("litellm.proxy.proxy_server.prisma_client", mock_client),
|
||||
):
|
||||
result: Final = await patch_user(user_id="user-mva", patch_ops=patch_ops)
|
||||
|
||||
update_data: Final = mock_db.litellm_usertable.update.call_args.kwargs["data"]
|
||||
metadata_json: Final = update_data["metadata"]
|
||||
assert isinstance(metadata_json, str)
|
||||
metadata: Final = TypeAdapter(dict[str, object]).validate_json(metadata_json)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {
|
||||
"employeeNumber": "42",
|
||||
"department": "Eng",
|
||||
"costCenter": "CC-10",
|
||||
}
|
||||
assert not any(isinstance(key, str) and key.casefold().startswith("urn:") for key in metadata)
|
||||
assert result.enterprise_user == SCIMEnterpriseUser(
|
||||
employeeNumber="42",
|
||||
department="Eng",
|
||||
costCenter="CC-10",
|
||||
)
|
||||
|
||||
|
||||
def test_apply_patch_ops_replace_enterprise_urn_path_from_idp():
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="replace",
|
||||
path="urn:ietf:params:scim:schemas:extension:enterprise:2.0:User:department",
|
||||
value="Eng",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(existing_user=_user_with_metadata({}), patch_ops=patch_ops)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {"department": "Eng"}
|
||||
assert not any(isinstance(key, str) and key.casefold().startswith("urn:") for key in metadata)
|
||||
|
||||
|
||||
def test_apply_patch_ops_pathless_enterprise_manager_sub_attribute_merges():
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="replace",
|
||||
value={
|
||||
f"{SCIM_ENTERPRISE_USER_SCHEMA.upper()}:MANAGER.VALUE": "manager-1",
|
||||
},
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata(
|
||||
{SCIM_ENTERPRISE_METADATA_KEY: {"manager": {"displayName": "Engineering Manager"}}}
|
||||
),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {
|
||||
"manager": {"value": "manager-1", "displayName": "Engineering Manager"}
|
||||
}
|
||||
|
||||
|
||||
def test_apply_patch_ops_replace_enterprise_manager_normalizes_case_insensitive_keys():
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="replace",
|
||||
path=f"{SCIM_ENTERPRISE_USER_SCHEMA}:manager",
|
||||
value={"VALUE": "mgr-2"},
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata(
|
||||
{
|
||||
SCIM_ENTERPRISE_METADATA_KEY: {
|
||||
"employeeNumber": "42",
|
||||
"manager": {"value": "mgr-1", "displayName": "Boss"},
|
||||
}
|
||||
}
|
||||
),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {
|
||||
"employeeNumber": "42",
|
||||
"manager": {"value": "mgr-2", "displayName": "Boss"},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("path", "value"),
|
||||
(
|
||||
pytest.param(
|
||||
f"{SCIM_ENTERPRISE_USER_SCHEMA}:manager",
|
||||
{"displayName": "Q"},
|
||||
id="pathed",
|
||||
),
|
||||
pytest.param(
|
||||
None,
|
||||
{SCIM_ENTERPRISE_USER_SCHEMA: {"manager": {"displayName": "Q"}}},
|
||||
id="pathless",
|
||||
),
|
||||
pytest.param(
|
||||
None,
|
||||
{SCIM_ENTERPRISE_USER_SCHEMA: {"MANAGER": {"displayName": "Q"}}},
|
||||
id="pathless-uppercase-key",
|
||||
),
|
||||
),
|
||||
)
|
||||
def test_apply_patch_ops_replace_manager_object_preserves_unspecified_sub_attributes(
|
||||
path: str | None,
|
||||
value: object,
|
||||
) -> None:
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[SCIMPatchOperation(op="replace", path=path, value=value)]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata(
|
||||
{
|
||||
SCIM_ENTERPRISE_METADATA_KEY: {
|
||||
"department": "A",
|
||||
"manager": {"value": "m", "displayName": "B"},
|
||||
}
|
||||
}
|
||||
),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {
|
||||
"department": "A",
|
||||
"manager": {"value": "m", "displayName": "Q"},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("manager_path", "manager_value"),
|
||||
(
|
||||
pytest.param(
|
||||
f"{SCIM_ENTERPRISE_USER_SCHEMA}:manager",
|
||||
{"$ref": "new"},
|
||||
id="pathed",
|
||||
),
|
||||
pytest.param(
|
||||
None,
|
||||
{SCIM_ENTERPRISE_USER_SCHEMA: {"manager": {"$ref": "new"}}},
|
||||
id="pathless",
|
||||
),
|
||||
),
|
||||
)
|
||||
def test_apply_patch_ops_batched_manager_ref_update_replaces_previous_link(
|
||||
manager_path: str | None,
|
||||
manager_value: object,
|
||||
) -> None:
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="replace",
|
||||
path=f"{SCIM_ENTERPRISE_USER_SCHEMA}:department",
|
||||
value="B",
|
||||
),
|
||||
SCIMPatchOperation(
|
||||
op="replace",
|
||||
path=manager_path,
|
||||
value=manager_value,
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata(
|
||||
{
|
||||
SCIM_ENTERPRISE_METADATA_KEY: {
|
||||
"department": "A",
|
||||
"manager": {"value": "m", "$ref": "old"},
|
||||
}
|
||||
}
|
||||
),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {
|
||||
"department": "B",
|
||||
"manager": {"value": "m", "$ref": "new"},
|
||||
}
|
||||
|
||||
|
||||
def test_apply_patch_ops_add_enterprise_manager_merges_existing_sub_attributes():
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="add",
|
||||
path=f"{SCIM_ENTERPRISE_USER_SCHEMA}:manager",
|
||||
value={"displayName": "New Boss"},
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata(
|
||||
{
|
||||
SCIM_ENTERPRISE_METADATA_KEY: {
|
||||
"employeeNumber": "42",
|
||||
"manager": {
|
||||
"value": "mgr-1",
|
||||
"$ref": "https://idp.example/Users/mgr-1",
|
||||
},
|
||||
}
|
||||
}
|
||||
),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {
|
||||
"employeeNumber": "42",
|
||||
"manager": {
|
||||
"value": "mgr-1",
|
||||
"$ref": "https://idp.example/Users/mgr-1",
|
||||
"displayName": "New Boss",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_apply_patch_ops_pathless_add_enterprise_manager_merges_existing_sub_attributes():
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="add",
|
||||
value={
|
||||
SCIM_ENTERPRISE_USER_SCHEMA: {
|
||||
"manager": {"displayName": "New Boss"},
|
||||
}
|
||||
},
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata(
|
||||
{
|
||||
SCIM_ENTERPRISE_METADATA_KEY: {
|
||||
"employeeNumber": "42",
|
||||
"manager": {
|
||||
"value": "mgr-1",
|
||||
"$ref": "https://idp.example/Users/mgr-1",
|
||||
},
|
||||
}
|
||||
}
|
||||
),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {
|
||||
"employeeNumber": "42",
|
||||
"manager": {
|
||||
"value": "mgr-1",
|
||||
"$ref": "https://idp.example/Users/mgr-1",
|
||||
"displayName": "New Boss",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_apply_patch_ops_remove_enterprise_manager_sub_attribute_preserves_other_fields():
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="remove",
|
||||
path=f"{SCIM_ENTERPRISE_USER_SCHEMA}:manager.value",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata(
|
||||
{
|
||||
SCIM_ENTERPRISE_METADATA_KEY: {
|
||||
"employeeNumber": "42",
|
||||
"manager": {"value": "mgr-1", "displayName": "Boss"},
|
||||
}
|
||||
}
|
||||
),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {
|
||||
"employeeNumber": "42",
|
||||
"manager": {"displayName": "Boss"},
|
||||
}
|
||||
|
||||
|
||||
def test_apply_patch_ops_remove_enterprise_attribute_preserves_other_attributes():
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="remove",
|
||||
path=f"{SCIM_ENTERPRISE_USER_SCHEMA}:costCenter",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata({SCIM_ENTERPRISE_METADATA_KEY: {"costCenter": "CC-10", "department": "Eng"}}),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {"department": "Eng"}
|
||||
|
||||
|
||||
def test_apply_patch_ops_remove_last_enterprise_attribute_drops_metadata_key():
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="remove",
|
||||
path=f"{SCIM_ENTERPRISE_USER_SCHEMA}:employeeNumber",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata({SCIM_ENTERPRISE_METADATA_KEY: {"employeeNumber": "42"}}),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert SCIM_ENTERPRISE_METADATA_KEY not in metadata
|
||||
|
||||
|
||||
def test_apply_patch_ops_unknown_enterprise_attribute_preserves_existing_and_applies_other_ops() -> None:
|
||||
existing_enterprise: Final = {
|
||||
"employeeNumber": "42",
|
||||
"manager": {"value": "mgr-1", "displayName": "Boss"},
|
||||
}
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="replace",
|
||||
path=f"{SCIM_ENTERPRISE_USER_SCHEMA}:shoeSize",
|
||||
value="10",
|
||||
),
|
||||
SCIMPatchOperation(op="replace", path="displayName", value="Updated User"),
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata({SCIM_ENTERPRISE_METADATA_KEY: existing_enterprise}),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == existing_enterprise
|
||||
assert not any(isinstance(key, str) and key.casefold().startswith("urn:") for key in metadata)
|
||||
assert update_data["user_alias"] == "Updated User"
|
||||
|
||||
|
||||
def test_apply_patch_ops_display_name_only_preserves_enterprise_value_unchanged() -> None:
|
||||
existing_enterprise: Final = {"Department": "Eng", "manager": "mgr-1"}
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[SCIMPatchOperation(op="replace", path="displayName", value="Ada")]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata({SCIM_ENTERPRISE_METADATA_KEY: existing_enterprise}),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {"Department": "Eng", "manager": "mgr-1"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", ("manager.foo", "manager.a.b", "department.x"))
|
||||
def test_apply_patch_ops_unknown_enterprise_nested_attribute_preserves_existing(path: str) -> None:
|
||||
existing_enterprise: Final = {
|
||||
"department": "Engineering",
|
||||
"manager": {"value": "mgr-1", "displayName": "Boss"},
|
||||
}
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="replace",
|
||||
path=f"{SCIM_ENTERPRISE_USER_SCHEMA}:{path}",
|
||||
value="ignored",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata({SCIM_ENTERPRISE_METADATA_KEY: existing_enterprise}),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == existing_enterprise
|
||||
|
||||
|
||||
@pytest.mark.parametrize("manager_id", ("mgr-id", ""))
|
||||
def test_apply_patch_ops_add_enterprise_manager_string_replaces_existing_manager(manager_id: str) -> None:
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="add",
|
||||
path=f"{SCIM_ENTERPRISE_USER_SCHEMA}:manager",
|
||||
value=manager_id,
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata(
|
||||
{
|
||||
SCIM_ENTERPRISE_METADATA_KEY: {
|
||||
"employeeNumber": "42",
|
||||
"manager": {"value": "old", "displayName": "Old Boss"},
|
||||
}
|
||||
}
|
||||
),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {
|
||||
"employeeNumber": "42",
|
||||
"manager": {"value": manager_id},
|
||||
}
|
||||
|
||||
|
||||
def test_apply_patch_ops_pathless_add_enterprise_manager_string_replaces_existing_manager() -> None:
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="add",
|
||||
value={SCIM_ENTERPRISE_USER_SCHEMA: {"manager": "mgr-id"}},
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata(
|
||||
{
|
||||
SCIM_ENTERPRISE_METADATA_KEY: {
|
||||
"employeeNumber": "42",
|
||||
"manager": {"value": "old", "displayName": "Old Boss"},
|
||||
}
|
||||
}
|
||||
),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert metadata[SCIM_ENTERPRISE_METADATA_KEY] == {
|
||||
"employeeNumber": "42",
|
||||
"manager": {"value": "mgr-id"},
|
||||
}
|
||||
|
||||
|
||||
def test_apply_patch_ops_remove_enterprise_manager_drops_manager() -> None:
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="remove",
|
||||
path=f"{SCIM_ENTERPRISE_USER_SCHEMA}:manager",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
update_data, _ = _apply_patch_ops(
|
||||
existing_user=_user_with_metadata(
|
||||
{SCIM_ENTERPRISE_METADATA_KEY: {"manager": {"value": "mgr-1", "displayName": "Boss"}}}
|
||||
),
|
||||
patch_ops=patch_ops,
|
||||
)
|
||||
|
||||
metadata: Final = update_data["metadata"]
|
||||
assert isinstance(metadata, dict)
|
||||
assert SCIM_ENTERPRISE_METADATA_KEY not in metadata
|
||||
|
||||
|
||||
def test_apply_patch_ops_pathless_enterprise_value_must_be_a_dict():
|
||||
patch_ops: Final = SCIMPatchOp(
|
||||
Operations=[
|
||||
SCIMPatchOperation(
|
||||
op="replace",
|
||||
value={SCIM_ENTERPRISE_USER_SCHEMA: "not-a-dict"},
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
_apply_patch_ops(existing_user=_user_with_metadata({}), patch_ops=patch_ops)
|
||||
|
||||
assert exc_info.value.status_code == 400
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue