From 78f4172da9da67fefee8088cf3f95ca20594e098 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=89loi=20Rivard?= Date: Sun, 27 Sep 2026 22:10:54 +0200 Subject: [PATCH 1/4] feat: read the payloads under the policy of the provider --- pyproject.toml | 2 +- scim2_server/provider.py | 3 ++- tests/conftest.py | 7 ++--- tests/integration/test_scim_application.py | 31 ++++++++++++++++++++++ uv.lock | 14 +++++----- 5 files changed, 45 insertions(+), 12 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d42c6a0..a875ec4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,7 +30,7 @@ classifiers = [ requires-python = ">= 3.11" dependencies = [ "scim2-filter-parser>=0.7.0", - "scim2-models>=0.8.2", + "scim2-models>=0.9.0", "werkzeug>=3.0.3", ] diff --git a/scim2_server/provider.py b/scim2_server/provider.py index cc8e959..8d23f5e 100644 --- a/scim2_server/provider.py +++ b/scim2_server/provider.py @@ -560,7 +560,8 @@ def wsgi_app(self, request: Request, environ): self.check_auth(request) # Wrap the entire call in a transaction. Should probably be optimized (use transaction only when necessary). - with self.backend: + # The provider makes its policy the one every payload is read under. + with self.provider, self.backend: response = getattr(self, f"call_{endpoint}")(request, **args) return response except RequestRedirect as e: diff --git a/tests/conftest.py b/tests/conftest.py index f3d9168..d3e9561 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -60,14 +60,15 @@ def wsgi(app): @pytest.fixture def wsgi_with(backend, scim_provider): - """Build clients of applications serving the default resources under another configuration.""" + """Build clients of applications serving the default resources under another configuration or policy.""" clients = [] - def build(config): + def build(config=None, policy=None): provider = ScimProvider( models=scim_provider.models, resource_types=scim_provider.resource_types, - config=config, + config=config or scim_provider.config, + policy=policy, ) transport = httpx2.WSGITransport(app=SCIMApplication(backend, provider)) client = httpx2.Client(transport=transport, base_url="https://scim.example.com") diff --git a/tests/integration/test_scim_application.py b/tests/integration/test_scim_application.py index 33e606d..9460ccd 100644 --- a/tests/integration/test_scim_application.py +++ b/tests/integration/test_scim_application.py @@ -8,6 +8,7 @@ from scim2_models import ETag from scim2_models import Filter from scim2_models import Patch +from scim2_models import ScimPolicy from scim2_models import ScimProvider from scim2_models import SearchRequest from scim2_models import ServiceProviderConfig @@ -1046,3 +1047,33 @@ def test_redirect(self, wsgi): r = wsgi.get("/v2", follow_redirects=False) assert r.is_redirect assert r.headers["Location"] == "https://scim.example.com/v2/" + + +class TestSCIMApplicationPolicy: + def test_undeclared_attribute_is_refused_by_default(self, wsgi): + """The default policy follows the strict reading and refuses what no schema declares.""" + r = wsgi.post( + "/v2/Users", json={"userName": "bjensen", "favoriteColor": "blue"} + ) + assert r.status_code == 400 + + def test_undeclared_attribute_follows_the_provider_policy( + self, wsgi_with, fake_user_data + ): + """Every payload the application reads follows the policy of its provider.""" + wsgi = wsgi_with(policy=ScimPolicy(unknown=ScimPolicy.Unknown.ignore)) + r = wsgi.post( + "/v2/Users", json={"userName": "bjensen", "favoriteColor": "blue"} + ) + assert r.status_code == 201 + assert "favoriteColor" not in r.json() + + r = wsgi.put( + f"/v2/Users/{r.json()['id']}", + json={"userName": "bjensen", "favoriteColor": "blue"}, + ) + assert r.status_code == 200 + assert "favoriteColor" not in r.json() + + r = wsgi.post("/v2/Users/.search", json={"count": 1, "sortOrder2": "x"}) + assert r.status_code == 200 diff --git a/uv.lock b/uv.lock index 9389be5..38cac7d 100644 --- a/uv.lock +++ b/uv.lock @@ -11,12 +11,12 @@ exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for exclude-newer-span = "P14D" [options.exclude-newer-package] -pytest-scim2-server = false +scim2-server = false +scim2-models = false scim2-cli = false scim2-client = false -scim2-models = false -scim2-server = false scim2-tester = false +pytest-scim2-server = false [[package]] name = "annotated-types" @@ -616,15 +616,15 @@ wheels = [ [[package]] name = "scim2-models" -version = "0.8.2" +version = "0.9.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "lark" }, { name = "pydantic", extra = ["email"] }, ] -sdist = { url = "https://files.pythonhosted.org/packages/d5/61/bf0746a7ab542c937b3888375f4df8857824a37833df7a5c3eb5ea7d0bd0/scim2_models-0.8.2.tar.gz", hash = "sha256:8ee4b69687d2680c06827af24a50c93a1d33fbfe0be492e0b2db34f4c5233ecd", size = 94332, upload-time = "2026-09-25T19:07:50.984Z" } +sdist = { url = "https://files.pythonhosted.org/packages/e6/25/a7722642f8420ecc0f9c4c271901b43208c263d64b1d7c0a1ed76cdfa661/scim2_models-0.9.0.tar.gz", hash = "sha256:a0b35178a2c03ca0b59300ce3c7724857fb9049dbf7f3b106b6696d32d0ae3cb", size = 96051, upload-time = "2026-09-27T20:00:15.438Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/e0/8e/6a793d8f2e2cb7f1f180f277b47ca104dbf87a14b8d53abd4e7f3ae37db9/scim2_models-0.8.2-py3-none-any.whl", hash = "sha256:b27eec4847cfdc57e173f78e81004932e7d7298e3192d05f5fe1ec579f4735e9", size = 116865, upload-time = "2026-09-25T19:07:49.035Z" }, + { url = "https://files.pythonhosted.org/packages/3b/f9/59cf15d896064e50e1a7b2f7570c4bc5fbe1071816d29f959456487ad617/scim2_models-0.9.0-py3-none-any.whl", hash = "sha256:934d11e7386edd2e0a682dfb279915c74d9f7b3027daf5333794a30bb7f813eb", size = 118555, upload-time = "2026-09-27T20:00:13.633Z" }, ] [[package]] @@ -650,7 +650,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "scim2-filter-parser", specifier = ">=0.7.0" }, - { name = "scim2-models", specifier = ">=0.8.2" }, + { name = "scim2-models", specifier = ">=0.9.0" }, { name = "werkzeug", specifier = ">=3.0.3" }, ] From 0911db3296e9352ae75797540c3273720321cc0d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=89loi=20Rivard?= Date: Sun, 27 Sep 2026 22:19:12 +0200 Subject: [PATCH 2/4] feat!: apply PATCH requests through scim2-models --- scim2_server/provider.py | 20 +-------- tests/integration/test_ms_entra.py | 19 ++++++-- tests/integration/test_scim_application.py | 52 ++++++++++++++++++++++ 3 files changed, 69 insertions(+), 22 deletions(-) diff --git a/scim2_server/provider.py b/scim2_server/provider.py index 8d23f5e..25b4259 100644 --- a/scim2_server/provider.py +++ b/scim2_server/provider.py @@ -37,7 +37,6 @@ from werkzeug.routing.exceptions import RequestRedirect from scim2_server.backend import Backend -from scim2_server.operators import patch_resource from scim2_server.utils import load_default_service_provider_config SEARCH_REQUEST_PARAMETERS = ( @@ -167,11 +166,6 @@ def etag_header(resource: Resource) -> dict[str, str]: """Return the ETag header of a published resource, if it has a version.""" return {"ETag": resource.meta.version} if resource.meta.version else {} - def apply_patch_operation(self, resource: Resource, patch_operation): - """Apply a PATCH operation to a resource.""" - for op in patch_operation.operations: - patch_resource(resource, op) - def check_preconditions(self, request: Request, resource: Resource) -> bool: """Evaluate the "If-Match" and "If-None-Match" headers against a resource. @@ -257,18 +251,8 @@ def call_single_resource( ) case _: # "PATCH" self.ensure_supported(self.config.patch, "PATCH") - payload = request.json - # MS Entra sometimes passes a "id" attribute - if "id" in payload: - del payload["id"] - operations = payload.get("Operations", []) - for operation in operations: - if "name" in operation: - # MS Entra sometimes passes a "name" attribute - del operation["name"] - ResourceModel = self.get_model(resource_type) - patch_operation = PatchOp[ResourceModel].model_validate(payload) + patch_operation = PatchOp[ResourceModel].model_validate(request.json) response_parameters = self.get_response_parameters( request, ResourceModel ) @@ -277,7 +261,7 @@ def call_single_resource( raise NotFound self.check_preconditions(request, resource) - self.apply_patch_operation(resource, patch_operation) + patch_operation.patch(resource) updated = self.backend.update_resource(resource_type, resource) if ( diff --git a/tests/integration/test_ms_entra.py b/tests/integration/test_ms_entra.py index e2276ba..976f01d 100644 --- a/tests/integration/test_ms_entra.py +++ b/tests/integration/test_ms_entra.py @@ -1,6 +1,13 @@ import uuid import pytest +from scim2_models import ScimPolicy + + +@pytest.fixture +def wsgi(wsgi_with): + """Serve the default resources under the tolerances Entra needs.""" + return wsgi_with(policy=ScimPolicy(unknown=ScimPolicy.Unknown.ignore)) class TestSCIMApplicationMSEntraIntegration: @@ -333,7 +340,9 @@ def test_groups(self, wsgi): r = wsgi.get(f"/v2/Groups/{group_id}") assert r.status_code == 200 assert r.json()["id"] == group_id - assert "new User" in r.text + assert id4 in r.text + # Members declare "display", not "displayName", which the policy ignores. + assert "new User" not in r.text # Patch remove all users r = wsgi.patch( @@ -1291,7 +1300,7 @@ def test_groups_with_garbage(self, wsgi): assert r.status_code == 201 group1 = r.json()["id"] - # Group patch add member + # Group patch add member, a bare string being no member value r = wsgi.patch( f"/v2/Groups/{group1}", json={ @@ -1302,7 +1311,8 @@ def test_groups_with_garbage(self, wsgi): "schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"], }, ) - assert r.status_code == 204 + assert r.status_code == 400 + assert r.json()["scimType"] == "invalidValue" # Group patch add member2 r = wsgi.patch( @@ -1315,7 +1325,8 @@ def test_groups_with_garbage(self, wsgi): "schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"], }, ) - assert r.status_code == 204 + assert r.status_code == 400 + assert r.json()["scimType"] == "invalidValue" # Get group r = wsgi.get(f"/v2/Groups/{group1}") diff --git a/tests/integration/test_scim_application.py b/tests/integration/test_scim_application.py index 9460ccd..d632cde 100644 --- a/tests/integration/test_scim_application.py +++ b/tests/integration/test_scim_application.py @@ -1077,3 +1077,55 @@ def test_undeclared_attribute_follows_the_provider_policy( r = wsgi.post("/v2/Users/.search", json={"count": 1, "sortOrder2": "x"}) assert r.status_code == 200 + + def test_patch_message_fields_follow_the_provider_policy(self, wsgi_with): + """Entra sends an "id" and an operation "name" that only a tolerant policy accepts.""" + payload = { + "schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + "id": "whatever", + "Operations": [ + {"name": "addMember", "op": "replace", "path": "active", "value": True} + ], + } + + wsgi = wsgi_with() + user_id = wsgi.post("/v2/Users", json={"userName": "bjensen"}).json()["id"] + r = wsgi.patch(f"/v2/Users/{user_id}", json=payload) + assert r.status_code == 400 + + wsgi = wsgi_with(policy=ScimPolicy(unknown=ScimPolicy.Unknown.ignore)) + user_id = wsgi.post("/v2/Users", json={"userName": "alice"}).json()["id"] + r = wsgi.patch(f"/v2/Users/{user_id}", json=payload) + assert r.status_code == 204 + assert wsgi.get(f"/v2/Users/{user_id}").json()["active"] is True + + def test_patch_applies_under_the_provider_policy(self, wsgi_with): + """A path filter matching no value creates the entry only when the policy allows it.""" + payload = { + "schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + "Operations": [ + { + "op": "replace", + "path": 'emails[type eq "work"].value', + "value": "bjensen@example.com", + } + ], + } + + wsgi = wsgi_with() + user_id = wsgi.post("/v2/Users", json={"userName": "bjensen"}).json()["id"] + r = wsgi.patch(f"/v2/Users/{user_id}", json=payload) + assert r.status_code == 400 + assert r.json()["scimType"] == "noTarget" + + wsgi = wsgi_with( + policy=ScimPolicy( + unmatched_path_filter=ScimPolicy.UnmatchedPathFilter.create + ) + ) + user_id = wsgi.post("/v2/Users", json={"userName": "alice"}).json()["id"] + r = wsgi.patch(f"/v2/Users/{user_id}", json=payload) + assert r.status_code == 204 + assert wsgi.get(f"/v2/Users/{user_id}").json()["emails"] == [ + {"type": "work", "value": "bjensen@example.com"} + ] From c39e8533e903ea31831b63b8d5842926dd9291d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=89loi=20Rivard?= Date: Mon, 28 Sep 2026 09:23:16 +0200 Subject: [PATCH 3/4] refactor!: remove the homegrown PATCH operators and filter evaluation --- README.md | 2 +- pyproject.toml | 1 - scim2_server/filter.py | 127 --------- scim2_server/operators.py | 387 -------------------------- scim2_server/utils.py | 136 --------- tests/test_operators.py | 568 -------------------------------------- tests/test_patch.py | 225 --------------- tests/test_utils.py | 362 ------------------------ uv.lock | 23 -- 9 files changed, 1 insertion(+), 1830 deletions(-) delete mode 100644 scim2_server/filter.py delete mode 100644 scim2_server/operators.py delete mode 100644 tests/test_operators.py delete mode 100644 tests/test_patch.py diff --git a/README.md b/README.md index ebffaa2..9492fa8 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # scim2-server This is an example WSGI-SCIM server using [scim2-models](https://github.com/python-scim/scim2-models). -It utilizes [werkzeug](https://werkzeug.palletsprojects.com/) and [scim2-filter-parser](https://github.com/15five/scim2-filter-parser) and keeps all resources in-memory, +It utilizes [werkzeug](https://werkzeug.palletsprojects.com/) and keeps all resources in-memory, they are lost once the process exits. ## Features diff --git a/pyproject.toml b/pyproject.toml index a875ec4..27a6a9c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,6 @@ classifiers = [ requires-python = ">= 3.11" dependencies = [ - "scim2-filter-parser>=0.7.0", "scim2-models>=0.9.0", "werkzeug>=3.0.3", ] diff --git a/scim2_server/filter.py b/scim2_server/filter.py deleted file mode 100644 index 5315dab..0000000 --- a/scim2_server/filter.py +++ /dev/null @@ -1,127 +0,0 @@ -from types import NoneType - -from scim2_filter_parser import ast as scim2ast -from scim2_models import BaseModel -from scim2_models import CaseExact -from scim2_models import InvalidFilterException - -from scim2_server.utils import get_by_alias -from scim2_server.utils import parse_new_value - - -def evaluate_filter( - obj: BaseModel | list[BaseModel], tree: scim2ast.AST -) -> bool | list[bool]: - """Resolve filters. - - This implementation is limited by the specifics of the scim2_filter_parser module. - - It works well enough for simple cases, though. It should be re- - implemented in the future. Probably once - https://github.com/python-scim/scim2-models/issues/17 - is implemented. - """ - from scim2_server.operators import ResolveOperator - from scim2_server.operators import ResolveResult - - match type(tree): - case scim2ast.Filter: - if tree.namespace is not None: - obj = ResolveOperator(tree.namespace.attr_name)(obj).get_values() - if isinstance(obj, list): - return [ - o - for o in obj - if bool(evaluate_filter(o, tree.expr)) != tree.negated - ] - return bool(evaluate_filter(obj, tree.expr)) != tree.negated - case scim2ast.LogExpr: - match tree.op: - case "and": - return evaluate_filter(obj, tree.expr1) and evaluate_filter( - obj, tree.expr2 - ) - case _: # "or" - return evaluate_filter(obj, tree.expr1) or evaluate_filter( - obj, tree.expr2 - ) - case _: # scim2ast.AttrExpr - path = tree.attr_path.attr_name - sub_attribute_name = None - if isinstance(path, scim2ast.Filter): - resolved = evaluate_filter(obj, path) - model = resolved[0] - attribute_name = "" - - # FIXME: Best guesses since there is no way to know for sure by this point - case_sensitivity = CaseExact.false - sub_attribute_name = path.namespace.sub_attr.value - else: - if tree.attr_path.uri: - path = tree.attr_path.uri + ":" + path - if tree.attr_path.sub_attr: - path += "." + tree.attr_path.sub_attr.value - resolved = ResolveOperator(path)(obj) - case_sensitivity = resolved.get_field_annotation(CaseExact) - model = resolved.model - attribute_name = resolved.attribute - - if isinstance(resolved, ResolveResult): - value = resolved.get_values() - else: - value = [ - getattr(v, get_by_alias(type(v), sub_attribute_name)) - for v in resolved - ] - - compare_value = None - if tree.comp_value: - if attribute_name: - compare_value = parse_new_value( - model, attribute_name, tree.comp_value.value - ) - else: - compare_value = tree.comp_value.value - if not case_sensitivity and isinstance(value, str): - value = value.lower() - if compare_value: - compare_value = compare_value.lower() - - match tree.value: - case "eq": - return value == compare_value - case "ne": - return value != compare_value - case "sw": - return value.startswith(compare_value) - case "ew": - return value.endswith(compare_value) - case "pr": - return bool(value) - case "co": - if value is None: - return False - return compare_value in value - case "gt": - check_comparable_value(value) - return value > compare_value - case "lt": - check_comparable_value(value) - return value < compare_value - case "ge": - check_comparable_value(value) - return value >= compare_value - case _: # "le" - check_comparable_value(value) - return value <= compare_value - return False - - -def check_comparable_value(value): - """Certain values may not be compared in a filter, see RFC 7644, section 3.4.2.2. - - "Boolean and Binary attributes SHALL cause a failed response (HTTP - status code 400) with "scimType" of "invalidFilter"." - """ - if isinstance(value, bytes | bool | NoneType): - raise InvalidFilterException() diff --git a/scim2_server/operators.py b/scim2_server/operators.py deleted file mode 100644 index 706d21d..0000000 --- a/scim2_server/operators.py +++ /dev/null @@ -1,387 +0,0 @@ -import re -from typing import Any - -from scim2_filter_parser.lexer import SCIMLexer -from scim2_filter_parser.parser import SCIMParser -from scim2_models import BaseModel -from scim2_models import InvalidPathException -from scim2_models import InvalidValueException -from scim2_models import Mutability -from scim2_models import MutabilityException -from scim2_models import NoTargetException -from scim2_models import PatchOperation -from scim2_models import Required -from scim2_models import Resource -from scim2_models import Returned -from scim2_models import SensitiveException - -from scim2_server.filter import evaluate_filter -from scim2_server.utils import get_by_alias -from scim2_server.utils import get_or_create -from scim2_server.utils import handle_extension -from scim2_server.utils import parse_new_value - -ATTRIBUTE_PATH_REGEX = re.compile( - r"^(?P\w+)(\[(?P.*)\])?(\.(?P\w+))?$" -) - - -def patch_resource(resource: Resource, operation: PatchOperation): - """Run a patch operation against a resource.""" - path = str(operation.path) if operation.path else None - match operation.op: - case PatchOperation.Op.add: - operator = AddOperator(path, operation.value) - operator(resource) - case PatchOperation.Op.remove: - operator = RemoveOperator(path, None) - operator(resource) - case _: # PatchOperation.Op.replace - operator = ReplaceOperator(path, operation.value) - operator(resource) - - -def parse_attribute_path(attribute_path: str | None) -> dict[str, Any] | None: - """Parse an attribute path and returns a dictionary of attributes. - - The attributes are the named captures in the regex - ATTRIBUTE_PATH_REGEX. - """ - if not attribute_path: - return {} - - if not hasattr(parse_attribute_path, "cache"): - parse_attribute_path.cache = {} - if attribute_path in parse_attribute_path.cache: - return parse_attribute_path.cache[attribute_path] - - match = ATTRIBUTE_PATH_REGEX.match(attribute_path) - if not match: - raise InvalidPathException() - parse_attribute_path.cache[attribute_path] = match.groupdict() - return match.groupdict() - - -class Operator: - """An operator operates on a resource and is constructed using a path and a value.""" - - OPERATE_ON_ROOT = True # Whether the operator may operate on the root of a resource (e.g. remove may not) - REQUIRES_VALUE = True # Whether the operator modifies the resource or not - RETURNS_VALUE = False # Whether the operator returns a result or not - - def __init__(self, path: str | None, value: Any | None): - self.path = path - self.value = value - - @classmethod - def init_return( - cls, model: BaseModel, attribute: str, sub_attribute: str | None, value: Any - ): - """Initialize the return value if the operator returns something.""" - pass - - def do_return(self): - """Return the return value for the operator.""" - return None - - @classmethod - def operation( - cls, model: BaseModel, attribute: str, value: Any, index: int | None = None - ): - """Perform the actual operation of the operator.""" - raise NotImplementedError - - def parse_path(self, model: BaseModel): - """Parse a path and optionally handles model extensions. - - :return: A tuple of the model to operate on and the parsed path. - """ - path = self.path - if isinstance(model, Resource): - model, path = handle_extension(model, self.path) - return model, parse_attribute_path(path) - - def __call__(self, model: BaseModel): - """Execute the operator against a model.""" - if not self.path: - self.call_on_root(model) - return self.do_return() - - model, path = self.parse_path(model) - - match path: - case { - "attribute": attribute, - "condition": None, - "sub_attribute": None, - }: - self.match_attribute(attribute, model) - return self.do_return() - case { - "attribute": attribute, - "condition": None, - "sub_attribute": sub_path, - }: - self.match_complex_attribute(attribute, model, sub_path) - return self.do_return() - case { - "attribute": attribute, - "condition": condition, - "sub_attribute": None, - }: - self.match_multi_valued_attribute(attribute, condition, model) - return self.do_return() - case { - "attribute": attribute, - "condition": condition, - "sub_attribute": sub_attribute, - }: - self.match_multi_valued_attribute_sub( - attribute, condition, model, sub_attribute - ) - return self.do_return() - case _: - self.call_on_root(model) - return self.do_return() - - def match_multi_valued_attribute_sub( - self, attribute: str, condition: str, model: BaseModel, sub_attribute: str - ): - attribute_name = get_by_alias(type(model), attribute) - multi_valued_attribute = get_or_create(model, attribute_name, True) - if not isinstance(multi_valued_attribute, list): - raise InvalidPathException() - token_stream = SCIMLexer().tokenize(condition) - condition = SCIMParser().parse(token_stream) - self.init_return(model, attribute_name, sub_attribute, self.value) - for value in multi_valued_attribute: - if evaluate_filter(value, condition): - self.operation(value, sub_attribute, self.value) - - def match_multi_valued_attribute( - self, attribute: str, condition: str, model: BaseModel - ): - if self.REQUIRES_VALUE and not isinstance(self.value, dict): - raise InvalidValueException() - attribute_name = get_by_alias(type(model), attribute) - multi_valued_attribute = get_or_create( - model, attribute_name, self.REQUIRES_VALUE - ) - if not isinstance(multi_valued_attribute, list): - raise InvalidPathException() - token_stream = SCIMLexer().tokenize(condition) - condition = SCIMParser().parse(token_stream) - if self.REQUIRES_VALUE: - for value in multi_valued_attribute: - if evaluate_filter(value, condition): - for k, v in self.value.items(): - self.operation(value, k, v) - elif self.RETURNS_VALUE: - self.init_return(model, attribute_name, None, self.value) - for index, value in enumerate(multi_valued_attribute): - if evaluate_filter(value, condition): - self.operation(model, attribute_name, self.value, index) - else: - new_value = [ - value - for value in multi_valued_attribute - if not evaluate_filter(value, condition) - ] - setattr(model, attribute_name, new_value) - - def match_complex_attribute(self, attribute: str, model: BaseModel, sub_path: str): - complex_attribute = get_or_create( - model, get_by_alias(type(model), attribute), self.REQUIRES_VALUE - ) - if isinstance(complex_attribute, list) and complex_attribute: - for value in complex_attribute: - self.match_attribute(sub_path, value) - else: - if not isinstance(complex_attribute, BaseModel): - raise InvalidPathException() - self.match_attribute(sub_path, complex_attribute) - - def match_attribute(self, attribute: str, model: BaseModel): - self.init_return(model, attribute, None, self.value) - self.operation(model, attribute, self.value) - - def call_on_root(self, model: Resource): - if not self.OPERATE_ON_ROOT: - raise NoTargetException() - if not isinstance(self.value, dict): - raise InvalidValueException() - for k, v in self.value.items(): - ext, scim_name = handle_extension(model, k) - if ext == model: - self.operation(model, scim_name, v) - else: - type(self)(self.path, v)(ext) - - -class AddOperator(Operator): - """The implementation for the PATCH "add" operator.""" - - @classmethod - def operation(cls, model: BaseModel, attribute: str, value: Any): - alias = get_by_alias(type(model), attribute) - if model.get_field_multiplicity(alias) and isinstance(value, list): - for v in value: - cls.operation(model, attribute, v) - return - - existing_value = getattr(model, alias) - new_value = parse_new_value(model, alias, value) - if new_value == existing_value: - return - - if model.get_field_annotation(alias, Mutability) == Mutability.read_only: - raise MutabilityException() - - if model.get_field_multiplicity(alias): - if getattr(model, alias) is None: - setattr(model, alias, []) - if getattr(new_value, "primary", False): - for value in getattr(model, alias): - value.primary = False - getattr(model, alias).append(new_value) - else: - if ( - model.get_field_annotation(alias, Required) == Required.true - and not new_value - ): - raise InvalidValueException() - setattr(model, alias, new_value) - - -class RemoveOperator(Operator): - """The implementation for the PATCH "remove" operator.""" - - OPERATE_ON_ROOT = False - REQUIRES_VALUE = False - - @classmethod - def operation(cls, model: BaseModel, attribute: str, value: Any): - alias = get_by_alias(type(model), attribute) - existing_value = getattr(model, alias) - if not existing_value: - return - - if model.get_field_annotation(alias, Mutability) in ( - Mutability.read_only, - Mutability.immutable, - ): - raise MutabilityException() - - if model.get_field_annotation(alias, Required) == Required.true: - raise InvalidValueException() - - setattr(model, alias, None) - - -class ReplaceOperator(Operator): - """The implementation for the PATCH "replace" operator.""" - - @classmethod - def operation(cls, model: BaseModel, attribute: str, value: Any): - alias = get_by_alias(type(model), attribute) - if model.get_field_multiplicity(alias) and not isinstance(value, list): - raise InvalidValueException() - - existing_value = getattr(model, alias) - new_value = parse_new_value(model, alias, value) - if new_value == existing_value: - return - - if model.get_field_annotation(alias, Mutability) == Mutability.read_only: - raise MutabilityException() - - if ( - model.get_field_annotation(alias, Required) == Required.true - and not new_value - ): - raise InvalidValueException() - setattr(model, alias, new_value) - - -class ResolveResult: - """A descriptor for the result returned from the "ResolveOperator", used to resolve attributes from a model.""" - - def __init__(self): - self.records = [] - self.result_idx = 0 - self.model = None - self.attribute = None - self.sub_attribute = None - - def add_result(self, model: BaseModel, attribute_name: str): - """Add a result to the descriptor.""" - self.records.append((model, attribute_name)) - - def add_result_index(self, model: BaseModel, attribute_name: str, index: int): - """Add a result to the descriptor. - - The resulting attribute is part of a multi-valued attribute - described by its index. - """ - self.records.append((model, attribute_name, index)) - - def _evaluate_result(self, record: tuple[str, str] | tuple[str, str, int]): - if len(record) == 2: - return getattr(*record) - else: - return getattr(record[0], record[1])[record[2]] - - def get_field_annotation(self, annotation_type: type): - if not self.model: - return None - return self.model.get_field_annotation(self.attribute, annotation_type) - - def get_values(self): - if not self.records: - return None - if len(self.records) == 1: - return self._evaluate_result(self.records[0]) - return [self._evaluate_result(record) for record in self.records] - - -class ResolveOperator(Operator): - """An operator that resolves attribute values from a model.""" - - REQUIRES_VALUE = False - RETURNS_VALUE = True - - def __init__(self, path: str | None): - super().__init__(path, ResolveResult()) - - def do_return(self): - ret = self.value - self.value = ResolveResult() - return ret - - @classmethod - def init_return( - cls, - model: BaseModel, - attribute: str, - sub_attribute: str | None, - value: ResolveResult, - ): - alias = get_by_alias(type(model), attribute) - value.model = model - value.attribute = alias - value.sub_attribute = sub_attribute - if ( - model.get_field_annotation(alias, Mutability) == Mutability.write_only - or model.get_field_annotation(alias, Returned) == Returned.never - ): - raise SensitiveException() - - @classmethod - def operation( - cls, model: BaseModel, attribute: str, value: Any, index: int | None = None - ): - alias = get_by_alias(type(model), attribute) - if index is None: - value.add_result(model, alias) - else: - value.add_result_index(model, alias, index) diff --git a/scim2_server/utils.py b/scim2_server/utils.py index 92351bd..57c2543 100644 --- a/scim2_server/utils.py +++ b/scim2_server/utils.py @@ -1,15 +1,7 @@ -import datetime import importlib.resources import json from typing import Any -from pydantic import EmailStr -from pydantic import ValidationError -from scim2_models import BaseModel -from scim2_models import InvalidValueException -from scim2_models import Mutability -from scim2_models import MutabilityException -from scim2_models import NoTargetException from scim2_models import Resource from scim2_models import ResourceType from scim2_models import Schema @@ -58,131 +50,3 @@ def load_default_provider() -> ScimProvider: load_default_resource_types().values(), config=load_default_service_provider_config(), ) - - -def get_by_alias( - r: type[BaseModel], scim_name: str, allow_none: bool = False -) -> str | None: - """Return the pydantic attribute name for a BaseModel type and given SCIM attribute name. - - :param r: BaseModel type - :param scim_name: SCIM attribute name - :param allow_none: Allow returning None if attribute is not found - :return: pydantic attribute name - :raises NoTargetException: If no attribute is found and allow_none - is False - """ - try: - return next( - k - for k, v in r.model_fields.items() - if v.serialization_alias.lower() == scim_name.lower() - ) - except StopIteration as e: - if allow_none: - return None - raise NoTargetException() from e - - -def get_or_create( - model: BaseModel, attribute_name: str, check_mutability: bool = False -): - """Get or creates a complex attribute model for a given resource. - - :param model: The model - :param attribute_name: The attribute name - :param check_mutability: If True, validate that the attribute is - mutable - :return: A complex attribute model - :raises MutabilityException: If attribute is not mutable and - check_mutability is True - """ - if check_mutability: - if model.get_field_annotation(attribute_name, Mutability) in ( - Mutability.read_only, - Mutability.immutable, - ): - raise MutabilityException() - ret = getattr(model, attribute_name, None) - if not ret: - if model.get_field_multiplicity(attribute_name): - ret = [] - setattr(model, attribute_name, ret) - else: - field_root_type = model.get_field_root_type(attribute_name) - ret = field_root_type() - setattr(model, attribute_name, ret) - return ret - - -def handle_extension(resource: Resource, scim_name: str) -> tuple[BaseModel, str]: - default_schema = str(resource.__class__.__schema__).lower() - if scim_name.lower().startswith(default_schema): - scim_name = scim_name[len(default_schema) :].lstrip(":") - return resource, scim_name - - if isinstance(resource, Resource): - for extension_model in resource.get_extension_models(): - extension_prefix = extension_model.lower() - if scim_name.lower().startswith(extension_prefix): - scim_name = scim_name[len(extension_prefix) :] - scim_name = scim_name.lstrip(":") - if extension_model.lower() not in [s.lower() for s in resource.schemas]: - resource.schemas.append(extension_model) - ext = get_or_create( - resource, get_by_alias(type(resource), extension_model) - ) - return ext, scim_name - return resource, scim_name - - -def parse_value(field_root_type: type, value: Any) -> Any: - """Parse a PATCH value according to the target field root type.""" - if isinstance(value, dict): - if not hasattr(field_root_type, "model_fields"): - raise TypeError - - # Work around mixed display/displayName payloads emitted by MS Entra. - if ( - "display" not in value - and "display" in field_root_type.model_fields - and "displayName" in value - ): - value = value.copy() - value["display"] = value["displayName"] - del value["displayName"] - return field_root_type.model_validate(value) - - if field_root_type is bool and isinstance(value, str): - return not value.lower() == "false" - - if field_root_type is datetime.datetime and isinstance(value, str): - return datetime.datetime.fromisoformat(value) - - if field_root_type is EmailStr and isinstance(value, str): - return value - - if hasattr(field_root_type, "model_fields"): - primary_value = get_by_alias(field_root_type, "value", True) - if primary_value is not None: - return field_root_type(value=value) - raise TypeError - - return field_root_type(value) - - -def parse_new_value(model: BaseModel, attribute_name: str, value: Any) -> Any: - """Given a model and attribute name, attempt to parse a new value so that the type matches the type expected by the model. - - :raises InvalidValueException: If attribute can not be mapped to - the required type - """ - field_root_type = model.get_field_root_type(attribute_name) - try: - if isinstance(value, list): - new_value = [parse_value(field_root_type, v) for v in value] - else: - new_value = parse_value(field_root_type, value) - except (AttributeError, TypeError, ValueError, ValidationError) as e: - raise InvalidValueException() from e - return new_value diff --git a/tests/test_operators.py b/tests/test_operators.py deleted file mode 100644 index 96c1c5c..0000000 --- a/tests/test_operators.py +++ /dev/null @@ -1,568 +0,0 @@ -import base64 -import datetime - -import pytest -from scim2_models import URN -from scim2_models import BaseModel -from scim2_models import CaseExact -from scim2_models import Email -from scim2_models import EnterpriseUser -from scim2_models import GroupMembership -from scim2_models import InvalidPathException -from scim2_models import InvalidValueException -from scim2_models import MutabilityException -from scim2_models import Name -from scim2_models import NoTargetException -from scim2_models import Resource -from scim2_models import User -from scim2_models import X509Certificate - -from scim2_server.operators import AddOperator -from scim2_server.operators import RemoveOperator -from scim2_server.operators import ReplaceOperator -from scim2_server.operators import ResolveOperator -from scim2_server.operators import ResolveResult -from scim2_server.operators import parse_attribute_path - - -class TestOperators: - def test_path_resolving(self): - assert parse_attribute_path("") == {} - assert parse_attribute_path("a") == { - "attribute": "a", - "condition": None, - "sub_attribute": None, - } - assert parse_attribute_path("a.b") == { - "attribute": "a", - "condition": None, - "sub_attribute": "b", - } - assert parse_attribute_path("a[x eq 5]") == { - "attribute": "a", - "condition": "x eq 5", - "sub_attribute": None, - } - assert parse_attribute_path("a[x eq 5].b") == { - "attribute": "a", - "condition": "x eq 5", - "sub_attribute": "b", - } - with pytest.raises(InvalidPathException): - assert parse_attribute_path("%invalid$$path") - with pytest.raises(InvalidPathException): - assert parse_attribute_path(".a..b[x]") - with pytest.raises(InvalidPathException): - assert parse_attribute_path("\\x") - - def test_simple_add_operator_unset_single_valued(self): - u = User() - AddOperator.operation(u, "userName", "foo") - assert u.user_name == "foo" - - def test_simple_add_operator_single_valued(self): - u = User(user_name="foo") - AddOperator.operation(u, "userName", "bar") - assert u.user_name == "bar" - - def test_simple_add_operator_multi_valued(self): - u = User() - AddOperator.operation( - u, "emails", {"type": "work", "primary": True, "value": "work@example.com"} - ) - assert u.emails == [Email(type="work", value="work@example.com", primary=True)] - - def test_simple_add_operator_multi_valued_primary(self): - u = User() - AddOperator.operation( - u, "emails", {"type": "work", "primary": True, "value": "work@example.com"} - ) - AddOperator.operation( - u, "emails", {"type": "home", "primary": True, "value": "home@example.com"} - ) - assert u.emails == [ - Email(type="work", value="work@example.com", primary=False), - Email(type="home", value="home@example.com", primary=True), - ] - - def test_simple_add_operator_immutable(self): - u = User() - with pytest.raises(MutabilityException): - AddOperator.operation(u, "id", "123") - u.id = "123" - AddOperator.operation(u, "id", "123") - assert u.id == "123" - - def test_simple_add_operator_invalid_value(self): - u = User() - with pytest.raises(InvalidValueException): - AddOperator.operation(u, "userName", {}) - with pytest.raises(InvalidValueException): - AddOperator.operation(u, "emails", "abc") - with pytest.raises(InvalidValueException): - AddOperator.operation(u, "emails", {"foo": 123}) - - def test_simple_add_operator_bool_value_parsing(self): - u = User() - AddOperator.operation(u, "active", "True") - assert u.active - - AddOperator.operation(u, "active", 1) - assert u.active - - AddOperator.operation(u, "active", 0) - assert not u.active - - AddOperator.operation(u, "active", "False") - assert not u.active - - def test_add_operator_root_object(self): - u = User() - AddOperator( - None, - { - "userName": "foo", - "name": { - "formatted": "Mr. Foo", - }, - }, - )(u) - assert u.user_name == "foo" - assert u.name == Name(formatted="Mr. Foo") - - def test_add_operator_fully_qualified_attribute_name(self): - u = User() - AddOperator( - None, {"urn:ietf:params:scim:schemas:core:2.0:User:userName": "foo"} - )(u) - assert u.user_name == "foo" - - def test_add_operator_overwrite_required_attribute(self): - u = User(user_name="A") - with pytest.raises(InvalidValueException): - AddOperator("userName", "")(u) - - def test_add_operator_multi_valued_list(self): - u = User() - AddOperator( - None, - { - "emails": [ - {"type": "work", "primary": True, "value": "work@example.com"}, - {"type": "other", "primary": False, "value": "other@example.com"}, - ] - }, - )(u) - assert u.emails == [ - Email(type="work", value="work@example.com", primary=True), - Email(type="other", value="other@example.com", primary=False), - ] - - def test_add_operator_root_object_invalid_value(self): - u = User() - with pytest.raises(InvalidValueException): - AddOperator(None, "abc")(u) - with pytest.raises(InvalidValueException): - AddOperator(None, 1)(u) - with pytest.raises(InvalidValueException): - AddOperator(None, [1, 2, 3])(u) - - def test_add_operator_simple_attribute(self): - u = User() - AddOperator("userName", "foo")(u) - assert u.user_name == "foo" - - def test_add_operator_complex_attribute(self): - u = User() - AddOperator("name.formatted", "foo")(u) - assert u.name == Name(formatted="foo") - - AddOperator("name.givenName", "bar")(u) - assert u.name == Name(formatted="foo", givenName="bar") - - def test_add_operator_complex_attribute_invalid_path(self): - u = User() - with pytest.raises(InvalidValueException): - AddOperator('name[givenName eq "Foo"]', "foo")(u) - - with pytest.raises(InvalidPathException): - AddOperator('name[givenName eq "Foo"].formatted', "foo")(u) - - with pytest.raises(InvalidPathException): - AddOperator('name[givenName eq "Foo"]', {"formatted": "foo"})(u) - - def test_add_operator_multi_valued_attribute_invalid_path(self): - u = User() - with pytest.raises(InvalidPathException): - AddOperator("emails.value", "work@example.com")(u) - - def test_add_operator_multi_valued_attribute_mutability(self): - u = User() - with pytest.raises(MutabilityException): - AddOperator( - "groups", - { - "value": "x", - "type": "direct", - }, - )(u) - - def test_add_operator_multi_valued_complex_attribute(self): - u = User( - emails=[ - Email(value="work@example.com", primary=True), - Email(value="home@example.com", type="home", primary=False), - ] - ) - AddOperator('emails[value eq "work@example.com"]', {"type": "work"})(u) - assert u.emails == [ - Email(value="work@example.com", type="work", primary=True), - Email(value="home@example.com", type="home", primary=False), - ] - - def test_add_operator_multi_valued_complex_attribute_multiple_matches(self): - u = User( - emails=[ - Email(value="work@example.com", primary=True), - Email(value="home@example.com", type="home", primary=False), - ] - ) - AddOperator('emails[value ew "@example.com"]', {"type": "home"})(u) - assert u.emails == [ - Email(value="work@example.com", type="home", primary=True), - Email(value="home@example.com", type="home", primary=False), - ] - - def test_add_operator_multi_valued_complex_attribute_sub_attribute(self): - u = User( - emails=[ - Email(value="work@example.com", primary=True), - Email(value="home@example.com", type="home", primary=False), - ] - ) - AddOperator('emails[value eq "work@example.com"].type', "work")(u) - assert u.emails == [ - Email(value="work@example.com", type="work", primary=True), - Email(value="home@example.com", type="home", primary=False), - ] - - def test_add_operator_multi_valued_complex_attribute_sub_attribute_multiple_matches( - self, - ): - u = User( - emails=[ - Email(value="work@example.com", primary=False), - Email(value="home@example.com", type="home", primary=False), - ] - ) - AddOperator("emails[value pr].type", "home")(u) - assert u.emails == [ - Email(value="work@example.com", type="home", primary=False), - Email(value="home@example.com", type="home", primary=False), - ] - - def test_add_operator_multi_valued_complex_attribute_value_only(self): - u = User() - AddOperator("emails", "work@example.com")(u) - assert u.emails == [ - Email(value="work@example.com"), - ] - with pytest.raises(InvalidValueException): - AddOperator("name", "Mr. Foo")(u) - - def test_add_operator_binary_data(self): - u = User() - AddOperator("x509Certificates", {"value": base64.b64encode(b"1234567")})(u) - assert u.x509_certificates == [ - # https://github.com/python-scim/scim2-models/issues/31 - # Should be value=b"1234567" - X509Certificate(value=b"MTIzNDU2Nw==") - ] - - def test_add_operator_datetime(self): - class Foo(Resource): - __schema__ = URN("urn:example:2.0:Foo") - dt: datetime.datetime | None = None - - f = Foo() - AddOperator("dt", "2010-01-23T04:56:22Z")(f) - assert f.dt == datetime.datetime(2010, 1, 23, 4, 56, 22, tzinfo=datetime.UTC) - - def test_add_operator_extension_simple(self): - u = User[EnterpriseUser]() - AddOperator( - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User:employeeNumber", - "1234", - )(u) - assert u.schemas == [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ] - assert u.EnterpriseUser.employee_number == "1234" - - def test_add_operator_extension_complex(self): - u = User[EnterpriseUser]() - AddOperator( - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User:manager.value", - "1234", - )(u) - assert u.schemas == [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ] - assert u.EnterpriseUser.manager.value == "1234" - - def test_add_operator_extension_root(self): - u = User[EnterpriseUser]() - AddOperator( - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - {"employeeNumber": "1234"}, - )(u) - assert u.EnterpriseUser.employee_number == "1234" - - def test_add_operator_extension_path(self): - u = User[EnterpriseUser]() - AddOperator( - None, - { - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": { - "employeeNumber": "1234" - } - }, - )(u) - assert u.schemas == [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ] - assert u.EnterpriseUser.employee_number == "1234" - - def test_add_operator_extension_path_complex(self): - u = User[EnterpriseUser]() - AddOperator( - None, - { - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": { - "manager": { - "value": "1234", - "ref": "./26118915-6090-4610-87e4-49d8ca9f808d", - }, - "employeeNumber": "0000", - } - }, - )(u) - AddOperator( - None, - { - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": { - "employeeNumber": "5678" - } - }, - )(u) - assert u.schemas == [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ] - assert u.EnterpriseUser.employee_number == "5678" - assert u.EnterpriseUser.manager.value == "1234" - - def test_simple_replace_operator_unset_single_valued(self): - u = User() - ReplaceOperator.operation(u, "displayName", "newName") - assert u.display_name == "newName" - - def test_simple_replace_operator_single_valued(self): - u = User(display_name="Foo") - ReplaceOperator.operation(u, "displayName", "Bar") - assert u.display_name == "Bar" - - def test_simple_replace_operator_unset_multi_valued(self): - u = User() - ReplaceOperator.operation( - u, - "emails", - [{"type": "work", "primary": True, "value": "work@example.com"}], - ) - assert u.emails == [Email(type="work", value="work@example.com", primary=True)] - - def test_simple_replace_operator_multi_valued(self): - u = User( - emails=[ - Email(type="work", value="work@example.org", primary=False), - Email(type="home", value="home@example.org", primary=True), - ] - ) - ReplaceOperator.operation( - u, - "emails", - [{"type": "work", "primary": True, "value": "work@example.com"}], - ) - assert u.emails == [ - Email(type="work", value="work@example.com", primary=True), - ] - - def test_simple_replace_operator_multi_valued_invalid_value(self): - u = User(emails=[]) - with pytest.raises(InvalidValueException): - ReplaceOperator.operation( - u, - "emails", - {"type": "work", "primary": True, "value": "work@example.com"}, - ) - - def test_simple_replace_operator_immutable(self): - u = User() - with pytest.raises(MutabilityException): - ReplaceOperator.operation(u, "id", "123") - u.id = "123" - ReplaceOperator.operation(u, "id", "123") - assert u.id == "123" - - def test_replace_operator_root_object(self): - u = User(user_name="bar", name=Name(family_name="Foo")) - ReplaceOperator( - None, - { - "userName": "foo", - "name": { - "formatted": "Mr. Foo", - }, - }, - )(u) - assert u.user_name == "foo" - assert u.name == Name(formatted="Mr. Foo") - - def test_replace_operator_extension_root_object(self): - u = User[EnterpriseUser](user_name="bar") - u.EnterpriseUser = EnterpriseUser(employee_number="4321") - ReplaceOperator( - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - {"employeeNumber": "1234"}, - )(u) - assert u.EnterpriseUser.employee_number == "1234" - - def test_replace_operator_overwrite_required_attribute(self): - u = User(user_name="A") - with pytest.raises(InvalidValueException): - ReplaceOperator("userName", "")(u) - - def test_simple_remove_operator_unset_single_valued(self): - u = User() - RemoveOperator.operation(u, "userName", None) - assert u.user_name is None - - def test_simple_remove_operator_single_valued(self): - u = User(display_name="Foo") - RemoveOperator.operation(u, "displayName", None) - assert u.display_name is None - - def test_simple_remove_operator_multi_valued(self): - u = User( - emails=[Email(value="work@example.com"), Email(value="home@example.com")] - ) - RemoveOperator.operation(u, "emails", None) - assert u.emails is None - - def test_simple_remove_operator_immutable(self): - u = User(id="123", groups=[GroupMembership()]) - with pytest.raises(MutabilityException): - RemoveOperator.operation(u, "id", None) - with pytest.raises(MutabilityException): - RemoveOperator.operation(u, "groups", None) - - def test_remove_operator_root_object(self): - u = User() - with pytest.raises(NoTargetException): - RemoveOperator("", None)(u) - - def test_remove_operator_required_attribute(self): - u = User(user_name="A") - with pytest.raises(InvalidValueException): - RemoveOperator("userName", None)(u) - - def test_remove_operator_complex_attribute(self): - u = User(id="123", user_name="A", name=Name(formatted="Mr. Foo")) - RemoveOperator("name", None)(u) - assert u.name is None - - def test_remove_operator_complex_attribute_sub_attribute(self): - u = User( - id="123", user_name="A", name=Name(formatted="Mr. Foo", family_name="Foo") - ) - RemoveOperator("name.formatted", None)(u) - assert u.name == Name(family_name="Foo") - - def test_remove_operator_multi_valued_filter(self): - u = User( - emails=[ - Email(value="work@example.com", primary=False), - Email(value="home@example.com", type="home", primary=False), - ] - ) - RemoveOperator('emails[type eq "home"]', None)(u) - assert u.emails == [Email(value="work@example.com", primary=False)] - - def test_remove_operator_multi_valued_filter_sub_attribute(self): - u = User( - emails=[ - Email(value="work@example.com", primary=False), - Email(value="home@example.com", type="home", primary=False), - ] - ) - RemoveOperator('emails[type eq "home"].type', None)(u) - assert u.emails == [ - Email(value="work@example.com", primary=False), - Email(value="home@example.com", primary=False), - ] - - def test_remove_operator_extension_root(self): - u = User[EnterpriseUser]() - u.EnterpriseUser = EnterpriseUser(employee_number="123") - with pytest.raises(NoTargetException): - RemoveOperator( - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", None - )(u) - - def test_remove_operator_extension_simple_attribute(self): - u = User[EnterpriseUser]() - u.EnterpriseUser = EnterpriseUser(employee_number="123") - RemoveOperator( - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User:employeeNumber", - None, - )(u) - assert u.EnterpriseUser.employee_number is None - - def test_resolve_operator_result(self): - result = ResolveResult() - assert result.get_field_annotation(CaseExact) is None - - def _resolve_value(self, path: str, model: BaseModel): - result = ResolveOperator(path)(model) - return result.get_values() - - def test_resolve_operator_unset_simple_attribute(self): - u = User() - assert self._resolve_value("userName", u) is None - assert self._resolve_value("active", u) is None - with pytest.raises(NoTargetException): - self._resolve_value("invalidAttribute", u) - - def test_resolve_operator_simple_attribute(self): - u = User(user_name="ABC") - assert self._resolve_value("userName", u) == "ABC" - - def test_resolve_operator_complex_attribute(self): - u = User(name=Name(formatted="Mr. Foo")) - assert self._resolve_value("name", u) == Name(formatted="Mr. Foo") - assert self._resolve_value("name.formatted", u) == "Mr. Foo" - assert self._resolve_value("name.givenName", u) is None - - def test_resolve_operator_complex_multi_valued_attribute(self): - work_email = Email(value="work@example.com", primary=True) - home_email = Email(value="home@example.com", type="home", primary=False) - u = User(emails=[work_email, home_email]) - assert self._resolve_value("emails", u) == [work_email, home_email] - assert self._resolve_value('emails[type eq "home"]', u) == home_email - assert self._resolve_value('emails[type eq "other"]', u) is None - assert self._resolve_value("emails[type pr]", u) == home_email - assert self._resolve_value("emails[not (type pr)]", u) == work_email - assert self._resolve_value("emails[primary eq true]", u) == work_email diff --git a/tests/test_patch.py b/tests/test_patch.py deleted file mode 100644 index 9f51428..0000000 --- a/tests/test_patch.py +++ /dev/null @@ -1,225 +0,0 @@ -import pytest -from scim2_models import URN -from scim2_models import MutabilityException -from scim2_models import PatchOperation -from scim2_models.resources.resource import Resource - -from scim2_server.operators import patch_resource - - -class TestPatch: - def test_patch_operation_add_simple(self, app): - user = app.provider.model_for("User")(id="123") - patch_resource( - user, - PatchOperation( - op=PatchOperation.Op.add, - value={ - "userName": "Foo", - }, - ), - ) - assert user.model_dump() == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "id": "123", - "userName": "Foo", - } - patch_resource( - user, - PatchOperation( - op=PatchOperation.Op.add, - value={ - "userName": "Bar", - }, - ), - ) - assert user.model_dump() == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "id": "123", - "userName": "Bar", - } - with pytest.raises(MutabilityException): - patch_resource( - user, - PatchOperation( - op=PatchOperation.Op.add, - value={ - "id": "4", - }, - ), - ) - assert user.model_dump() == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "id": "123", - "userName": "Bar", - } - - def test_patch_operation_add_complex(self, app): - user = app.provider.model_for("User")(id="123") - patch_resource( - user, - PatchOperation( - op=PatchOperation.Op.add, - value={ - "name": {"formatted": "Mr. Foo"}, - }, - ), - ) - assert user.model_dump() == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "id": "123", - "name": { - "formatted": "Mr. Foo", - }, - } - patch_resource( - user, - PatchOperation( - op=PatchOperation.Op.add, - value={ - "name": {"givenName": "Baz"}, - }, - ), - ) - assert user.model_dump() == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "id": "123", - "name": { - "givenName": "Baz", - }, - } - patch_resource( - user, - PatchOperation( - op=PatchOperation.Op.add, - path="name", - value={ - "formatted": "Mr. Foo", - }, - ), - ) - assert user.model_dump() == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "id": "123", - "name": { - "formatted": "Mr. Foo", - }, - } - patch_resource( - user, - PatchOperation( - op=PatchOperation.Op.add, - path="name.familyName", - value="Jensen", - ), - ) - assert user.model_dump() == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "id": "123", - "name": { - "formatted": "Mr. Foo", - "familyName": "Jensen", - }, - } - - def test_patch_operation_add_multi_valued(self, app): - user = app.provider.model_for("User")(id="123") - patch_resource( - user, - PatchOperation( - op=PatchOperation.Op.add, - path="emails", - value={ - "value": "foo@example.com", - "type": "work", - "primary": True, - }, - ), - ) - assert user.model_dump() == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "id": "123", - "emails": [ - { - "value": "foo@example.com", - "type": "work", - "primary": True, - } - ], - } - patch_resource( - user, - PatchOperation( - op=PatchOperation.Op.add, - path="emails", - value={ - "value": "bar@example.com", - "type": "home", - "primary": True, - }, - ), - ) - assert user.model_dump() == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "id": "123", - "emails": [ - { - "value": "foo@example.com", - "type": "work", - "primary": False, - }, - { - "value": "bar@example.com", - "type": "home", - "primary": True, - }, - ], - } - - def test_patch_replace_multivalued_primitive_attribute(self): - """Replace a multi-valued primitive attribute.""" - - class MyResource(Resource): - __schema__ = URN("urn:example:schemas:MyResource") - - tags: list[str] | None = None - - resource = MyResource(id="123") - - patch_resource( - resource, - PatchOperation( - op=PatchOperation.Op.replace_, - path="tags", - value=["tag1", "tag2"], - ), - ) - - assert resource.tags == ["tag1", "tag2"] diff --git a/tests/test_utils.py b/tests/test_utils.py index 86b572f..80054bb 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,369 +1,7 @@ -import pytest -from scim2_filter_parser.lexer import SCIMLexer -from scim2_filter_parser.parser import SCIMParser -from scim2_models import Context -from scim2_models import EnterpriseUser -from scim2_models import InvalidFilterException -from scim2_models import Meta -from scim2_models import MutabilityException -from scim2_models import Name -from scim2_models import NoTargetException -from scim2_models import ResponseParameters -from scim2_models import SensitiveException -from scim2_models import User - -from scim2_server.filter import evaluate_filter -from scim2_server.operators import ResolveOperator -from scim2_server.utils import get_or_create from scim2_server.utils import load_default_provider from scim2_server.utils import load_default_schemas -class TestUtils: - def test_case_sensitivity(self): - u = User[EnterpriseUser].model_validate( - { - "userName": "abc", - "displayName": "Barbara Jensen", - "name": { - "formatted": "Barbara Jensen", - }, - "active": True, - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": { - "employeeNumber": "1234" - }, - } - ) - assert u.display_name == "Barbara Jensen" - - u.model_dump( - scim_ctx=Context.RESOURCE_QUERY_RESPONSE, - response_parameters=ResponseParameters( - attributes=[ - "displayname", - "urn:IETF:params:scim:schemas:core:2.0:User:userName", - "urn:IETF:params:scim:schemas:core:2.0:User:name.FORMATTED", - "acTIVe", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User:Employeenumber", - ] - ), - ) - - def test_match_filter(self, app): - user = app.provider.model_for("User").model_validate( - { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "userName": "ABC", - "name": {"formatted": "DEF"}, - "password": "(Rtk_Nuyz5", - "title": "Vice President", - "userType": "Employee", - "active": True, - "emails": [ - {"value": "foo@example.com", "type": "work"}, - {"value": "bar@example.com", "type": "home"}, - {"value": "home2@example.net", "type": "home"}, - ], - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": { - "employeeNumber": "X", - }, - "meta": { - "resourceType": "User", - "created": "2024-08-01T12:40:00Z", - "lastModified": "2024-08-02T15:15:00Z", - }, - }, - ) - - def evaluate(filter_str: str) -> bool: - token_stream = SCIMLexer().tokenize(filter_str) - tree = SCIMParser().parse(token_stream) - return evaluate_filter(user, tree) - - assert evaluate("userName pr") - assert not evaluate("not(userName pr)") - assert evaluate('userName eq "ABC"') - assert evaluate('urn:ietf:params:scim:schemas:core:2.0:User:userName eq "ABC"') - assert evaluate( - 'urn:ietf:params:scim:schemas:extension:enterprise:2.0:User:employeeNumber sw "X"' - ) - assert evaluate('userName co "B"') - assert evaluate('userName co "b"') - assert evaluate('userName gt "A"') - assert evaluate('userName ge "A"') - assert not evaluate('userName lt "A"') - assert not evaluate('userName le "A"') - assert evaluate('userName eq "A" or userName eq "ABC"') - assert not evaluate('not(userName gt "A")') - assert not evaluate('userName gt "B"') - assert evaluate('userName sw "ABC"') - assert evaluate('userName eq "abc"') - assert not evaluate('userName sw "C"') - assert evaluate('userName sw "a"') - assert evaluate('not (userName eq "DEF")') - assert evaluate('name.formatted eq "DEF"') - assert evaluate('name.formatted eq "def"') - assert evaluate('meta.created ge "2019-05-13T04:42:34Z"') - assert evaluate('meta.created gt "2019-05-13T04:42:34Z"') - assert not evaluate('meta.created le "2019-05-13T04:42:34Z"') - assert not evaluate('meta.created lt "2019-05-13T04:42:34Z"') - assert evaluate('meta.created eq "2024-08-01T12:40:00Z"') - assert evaluate('meta.created lt "2025-01-01T00:00:00Z"') - - with pytest.raises(SensitiveException): - # Password is sensitive (never returned), filter must not match - # to not reveal any information about the value - assert not evaluate('password sw "(Rtk_"') - - assert evaluate('emails[type eq "home"]') - assert evaluate('not (emails[type eq "other"])') - assert not evaluate("not (emails[value pr])") - assert not evaluate('emails[type eq "foo"]') - assert evaluate('emails[value pr].type co "home"') - assert not evaluate('emails[value pr].value co "x@example.com"') - - with pytest.raises(NoTargetException): - assert not evaluate('name[type eq "foo"]') - - with pytest.raises(InvalidFilterException): - assert not evaluate("active gt 5") - - with pytest.raises(InvalidFilterException): - assert not evaluate("active lt 5") - - with pytest.raises(InvalidFilterException): - assert not evaluate("active ge 5") - - with pytest.raises(InvalidFilterException): - assert not evaluate("active le 5") - - # Filters based on examples from RFC 7644, Section 3.4.2.2 - assert not evaluate('userName eq "bjensen"') - assert not evaluate('name.familyName co "O\'Malley"') - assert not evaluate('userName sw "J"') - assert not evaluate( - 'urn:ietf:params:scim:schemas:core:2.0:User:userName sw "J"' - ) - assert evaluate('urn:ietf:params:scim:schemas:core:2.0:User:userName sw "A"') - assert evaluate("title pr") - assert evaluate('meta.lastModified gt "2011-05-13T04:42:34Z"') - assert evaluate('meta.lastModified ge "2011-05-13T04:42:34Z"') - assert not evaluate('meta.lastModified lt "2011-05-13T04:42:34Z"') - assert not evaluate('meta.lastModified le "2011-05-13T04:42:34Z"') - assert evaluate('title pr and userType eq "Employee"') - assert evaluate('title pr or userType eq "Intern"') - assert not evaluate( - 'schemas eq "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"' - ) - assert not evaluate( - 'userType eq "Employee" and (emails co "example.com" or emails.value co "example.org")' - ) - assert evaluate( - 'userType eq "Employee" and (emails co "example.com" or emails.value co "bar@example.com")' - ) - assert not evaluate( - 'userType ne "Employee" and not (emails co "example.com" or emails.value co "example.org")' - ) - assert not evaluate('userType eq "Employee" and (emails.type eq "work")') - assert evaluate( - 'userType eq "Employee" and emails[type eq "work" and value co "@example.com"]' - ) - assert evaluate( - 'emails[type eq "work" and value co "@example.com"] or ims[type eq "xmpp" and value co "@foo.com"]' - ) - - def test_attribute_resolving(self, app): - user = app.provider.model_for("User").model_validate( - { - "schemas": [ - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - "urn:ietf:params:scim:schemas:core:2.0:User", - ], - "userName": "ABC", - "name": {"formatted": "DEF"}, - "emails": [ - {"value": "foo@example.com", "type": "work"}, - {"value": "bar@example.com", "type": "home"}, - {"value": "home2@example.net", "type": "home"}, - ], - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": { - "employeeNumber": "X", - }, - }, - scim_ctx=Context.RESOURCE_CREATION_REQUEST, - ) - - def validate( - path, - expected_value, - expected_attr_name, - expected_sub_attribute_name, - expected_class_name, - ): - result = ResolveOperator(path)(user) - value = result.get_values() - if expected_value is not None: - assert value == expected_value - assert expected_class_name in repr(result.model) - assert result.attribute == expected_attr_name - assert result.sub_attribute == expected_sub_attribute_name - return value - - validate( - "urn:ietf:params:scim:schemas:core:2.0:User:userName", - "ABC", - "user_name", - None, - "User[EnterpriseUser]", - ) - validate("userName", "ABC", "user_name", None, "User[EnterpriseUser]") - validate( - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User:employeeNumber", - "X", - "employee_number", - None, - "EnterpriseUser", - ) - emails = validate("emails", None, "emails", None, "User[EnterpriseUser]") - assert len(emails) == 3 - assert emails[0].value == "foo@example.com" - assert emails[0].type == "work" - - validate("name.formatted", "DEF", "formatted", None, "Name") - assert validate("name.givenName", None, "given_name", None, "Name") is None - - emails = validate( - 'emails[type eq "work"]', None, "emails", None, "User[EnterpriseUser]" - ) - assert emails.value == "foo@example.com" - assert emails.type == "work" - - emails_fully_qualified = validate( - 'urn:ietf:params:scim:schemas:core:2.0:User:emails[type eq "work"]', - None, - "emails", - None, - "User[EnterpriseUser]", - ) - assert emails == emails_fully_qualified - - validate( - 'emails[type eq "work"].value', - "foo@example.com", - "emails", - "value", - "Emails", - ) - - validate( - 'emails[type eq "work" or value ew "example.net"].value', - [ - "foo@example.com", - "home2@example.net", - ], - "emails", - "value", - "Emails", - ) - - validate( - 'emails[not (type eq "work") and value pr].value', - ["bar@example.com", "home2@example.net"], - "emails", - "value", - "Emails", - ) - - def test_dump_creation(self, app): - user = User(id="1", user_name="ABC") - user.name = Name(formatted="Barbara") - user.meta = Meta( - resource_type="User", - location="/v2/Users/foo", - ) - user.model_dump(scim_ctx=Context.RESOURCE_CREATION_RESPONSE) - - def test_dump_extension(self, app): - user = User[EnterpriseUser].model_validate( - { - "userName": "thomas38@harding-herman.com", - "displayName": "Misc. Michael Rodriguez", - "name": { - "givenName": "Michael", - "familyName": "Rodriguez", - "formatted": "Misc. Michael Rodriguez", - "honorificPrefix": "Misc.", - "honorificSuffix": "", - }, - "active": True, - "password": "&U+VPTt%(3", - "emails": [ - { - "value": "thomas38@harding-herman.com", - "type": "work", - "primary": True, - }, - { - "value": "christine08@gmail.com", - "type": "home", - "primary": False, - }, - ], - "phoneNumbers": [], - "addresses": [ - { - "streetAddress": "910 Armstrong Garden", - "locality": "Collinsmouth", - "postalCode": "70055", - "country": "Falkland Islands (Malvinas)", - "type": "work", - "formatted": "910 Armstrong Garden\nCollinsmouth\n70055 Falkland Islands (Malvinas)", - } - ], - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": { - "employeeNumber": "80612", - "organization": "Randolph Group", - }, - } - ) - assert user.model_dump( - scim_ctx=Context.RESOURCE_CREATION_RESPONSE, - response_parameters=ResponseParameters(attributes=["userName"]), - ) == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "userName": "thomas38@harding-herman.com", - } - assert user.model_dump( - scim_ctx=Context.RESOURCE_CREATION_RESPONSE, - response_parameters=ResponseParameters( - attributes=[ - "userName", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User:employeeNumber", - ] - ), - ) == { - "schemas": [ - "urn:ietf:params:scim:schemas:core:2.0:User", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", - ], - "userName": "thomas38@harding-herman.com", - "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": { - "employeeNumber": "80612", - }, - } - - def test_get_or_create_mutability(self): - u = User() - with pytest.raises(MutabilityException): - get_or_create(u, "groups", True) - - def test_the_default_provider_publishes_the_default_schemas(): """The schemas rebuilt from the default models are the ones the package ships.""" published = [schema.model_dump() for schema in load_default_provider().schemas] diff --git a/uv.lock b/uv.lock index 38cac7d..5fbcbad 100644 --- a/uv.lock +++ b/uv.lock @@ -602,18 +602,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/30/12/823d9a321904ccfd2969a24b84fdfd1e6614c707ec569c62879bf1dbc6c5/python_discovery-1.5.3-py3-none-any.whl", hash = "sha256:8305296358f1aa2ed302a25b84be7df84fef8ca47c7dce2da63cb7325333044e", size = 38290, upload-time = "2026-08-24T14:48:45.305Z" }, ] -[[package]] -name = "scim2-filter-parser" -version = "0.7.0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "sly" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/25/c0/2f4d5caee8faa8f0d0979584c4aab95fa69d626d6f9d211dd6bb7089bc2f/scim2_filter_parser-0.7.0.tar.gz", hash = "sha256:1e11dbe2e186fc1be6d93732b467a3bbaa9deff272dfeb3a0540394cfab7030c", size = 21358, upload-time = "2024-07-20T16:38:23.2Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/57/54/b54961bfc5018fa593758c439fe0d4a22fbadfabff49a7559850af9a79e1/scim2_filter_parser-0.7.0-py3-none-any.whl", hash = "sha256:a74f90a2d52a77e0f1bc4d77e84b79f88749469f6f7192d64a4f92e4fe50ab69", size = 23409, upload-time = "2024-07-20T16:38:21.525Z" }, -] - [[package]] name = "scim2-models" version = "0.9.0" @@ -632,7 +620,6 @@ name = "scim2-server" version = "0.2.0" source = { editable = "." } dependencies = [ - { name = "scim2-filter-parser" }, { name = "scim2-models" }, { name = "werkzeug" }, ] @@ -649,7 +636,6 @@ dev = [ [package.metadata] requires-dist = [ - { name = "scim2-filter-parser", specifier = ">=0.7.0" }, { name = "scim2-models", specifier = ">=0.9.0" }, { name = "werkzeug", specifier = ">=3.0.3" }, ] @@ -664,15 +650,6 @@ dev = [ { name = "tox-uv", specifier = ">=1.16.0" }, ] -[[package]] -name = "sly" -version = "0.5" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/41/8a/59e943f7b27904c7756a7b565ffbd55f3841f5cd3d2da2b2b0713c49e488/sly-0.5.tar.gz", hash = "sha256:251d42015e8507158aec2164f06035df4a82b0314ce6450f457d7125e7649024", size = 66702, upload-time = "2022-10-25T14:35:30.592Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/8e/4d/c96d807295183f2360329cd8d8bf5e8072c53d664125b3858c04153f026e/sly-0.5-py3-none-any.whl", hash = "sha256:20485483259eec7f6ba85ff4d2e96a4e50c6621902667fc2695cc8bc2a3e5133", size = 28864, upload-time = "2022-10-25T14:35:28.054Z" }, -] - [[package]] name = "time-machine" version = "3.5.0" From 04eb6fd9535f70a754e50b1db189a9840ddce831 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=89loi=20Rivard?= Date: Mon, 28 Sep 2026 10:05:57 +0200 Subject: [PATCH 4/4] fix: keep the ETag and lastModified when a PATCH changes nothing --- scim2_server/provider.py | 11 ++++++----- tests/integration/test_etags.py | 26 ++++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 5 deletions(-) diff --git a/scim2_server/provider.py b/scim2_server/provider.py index 25b4259..358734c 100644 --- a/scim2_server/provider.py +++ b/scim2_server/provider.py @@ -261,16 +261,17 @@ def call_single_resource( raise NotFound self.check_preconditions(request, resource) - patch_operation.patch(resource) - updated = self.backend.update_resource(resource_type, resource) + # A PATCH that changes nothing keeps meta.lastModified and the ETag. + if patch_operation.patch(resource): + resource = self.backend.update_resource(resource_type, resource) if ( response_parameters.attributes or response_parameters.excluded_attributes ): - updated = self.publish(request, updated) + resource = self.publish(request, resource) return self.make_response( - updated.model_dump( + resource.model_dump( scim_ctx=Context.RESOURCE_REPLACEMENT_RESPONSE, response_parameters=response_parameters, ) @@ -282,7 +283,7 @@ def call_single_resource( return self.make_response( None, 204, - headers=self.etag_header(self.publish(request, updated)), + headers=self.etag_header(self.publish(request, resource)), ) @staticmethod diff --git a/tests/integration/test_etags.py b/tests/integration/test_etags.py index b448752..cbe4941 100644 --- a/tests/integration/test_etags.py +++ b/tests/integration/test_etags.py @@ -161,6 +161,32 @@ def test_resource_patch_etag_match(self, wsgi, first_fake_user): ) assert r.json()["userName"] == "Foo" + def test_resource_patch_without_change_keeps_the_version( + self, wsgi, first_fake_user + ): + """A PATCH that changes nothing keeps meta.lastModified and the ETag (RFC 7643 §3.1).""" + initial = wsgi.get(f"/v2/Users/{first_fake_user}").json() + + r = wsgi.patch( + f"/v2/Users/{first_fake_user}", + json={ + "schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + "Operations": [ + { + "op": "replace", + "path": "userName", + "value": initial["userName"], + } + ], + }, + ) + assert r.status_code == 204 + assert r.headers["etag"] == initial["meta"]["version"] + + meta = wsgi.get(f"/v2/Users/{first_fake_user}").json()["meta"] + assert meta["version"] == initial["meta"]["version"] + assert meta["lastModified"] == initial["meta"]["lastModified"] + def test_resource_get_not_modified_carries_the_etag(self, wsgi, first_fake_user): """RFC 7232 §4.1: a 304 carries the ETag a 200 would have carried.""" version = wsgi.get(f"/v2/Users/{first_fake_user}").headers["etag"]