diff --git a/scim2_models/messages/patch_op.py b/scim2_models/messages/patch_op.py index a49f010..0e606bc 100644 --- a/scim2_models/messages/patch_op.py +++ b/scim2_models/messages/patch_op.py @@ -37,6 +37,19 @@ ResourceT = TypeVar("ResourceT", bound=Resource[Any]) +def _commit(resource: Any, working: Any) -> None: + """Write a patched copy back onto the resource the caller holds. + + Assignment is bypassed on purpose: ``working`` was built by the very passes + ``validate_assignment`` would run again. + """ + resource.__dict__.clear() + resource.__dict__.update(working.__dict__) + resource.__pydantic_fields_set__.clear() + resource.__pydantic_fields_set__.update(working.__pydantic_fields_set__) + resource.__pydantic_private__ = working.__pydantic_private__ + + def _targeted_attributes(value: Any) -> dict[str, Any]: """Return the attributes an operation without a path writes. @@ -559,6 +572,11 @@ def patch(self, resource: ResourceT, scim_policy: ScimPolicy | None = None) -> b ``primary`` sub-attribute to ``True``, any other values in the same multi-valued attribute will have their ``primary`` set to ``False`` automatically. + The operations are applied as a whole: when one fails, the resource is + left as it was. The resource object itself is kept, but the values it + holds are replaced, so a reference taken on one of them beforehand no + longer reflects the resource. + :param resource: The SCIM resource to patch. This object is modified in-place. :param scim_policy: The :class:`~scim2_models.ScimPolicy` the patch is applied under. Defaults to the strict reading of the specification. @@ -570,14 +588,20 @@ def patch(self, resource: ResourceT, scim_policy: ScimPolicy | None = None) -> b return False modified = False + # ยง3.5.2 has a failing operation leave the resource as it was, and an + # operation only fails once tried: a filter selecting nothing is known + # from the state, not from the payload. + working = resource.model_copy(deep=True) + # The policy is made ambient for the whole application: the passes it # governs below are revalidations that start from no call of ours. with _effective_policy(scim_policy): # RFC 7644 Section 3.5.2: "Apply each operation in sequence" for operation in self.operations: - if self._apply_operation(resource, operation): + if self._apply_operation(working, operation): modified = True + _commit(resource, working) return modified def _apply_operation( diff --git a/tests/test_patch_op_atomicity.py b/tests/test_patch_op_atomicity.py new file mode 100644 index 0000000..e133026 --- /dev/null +++ b/tests/test_patch_op_atomicity.py @@ -0,0 +1,184 @@ +"""The resource a failing patch was applied to is left as it was.""" + +from typing import Annotated + +import pytest + +from scim2_models import URN +from scim2_models import Email +from scim2_models import EnterpriseUser +from scim2_models import MutabilityException +from scim2_models import NoTargetException +from scim2_models import PatchOp +from scim2_models import PatchOperation +from scim2_models import ScimPolicy +from scim2_models import User +from scim2_models.annotations import Mutability +from scim2_models.resources.resource import Resource + + +def test_an_operation_failing_on_a_selection_undoes_the_ones_before_it(): + """A peer sends the operations it wants applied together, not one by one.""" + user = User(user_name="bjensen", title="Engineer") + patch = PatchOp[User]( + operations=[ + PatchOperation[User]( + op=PatchOperation.Op.replace_, path="title", value="Manager" + ), + PatchOperation[User]( + op=PatchOperation.Op.replace_, + path='emails[type eq "work"].value', + value="bjensen@example.com", + ), + ] + ) + + with pytest.raises(NoTargetException): + patch.patch(user) + + assert user.title == "Engineer" + + +def test_an_operation_failing_on_mutability_undoes_the_ones_before_it(): + """An immutable attribute already holding a value is only known from the state.""" + + class Dummy(Resource): + __schema__ = URN("urn:test:TestResource") + + mutable: str + immutable: Annotated[str, Mutability.immutable] + + resource = Dummy.model_construct(mutable="before", immutable="original") + patch = PatchOp[Dummy]( + operations=[ + PatchOperation[Dummy]( + op=PatchOperation.Op.replace_, path="mutable", value="after" + ), + PatchOperation[Dummy]( + op=PatchOperation.Op.replace_, path="immutable", value="new_value" + ), + ] + ) + + with pytest.raises(MutabilityException): + patch.patch(resource) + + assert resource.mutable == "before" + assert resource.immutable == "original" + + +def test_an_operation_failing_on_primary_undoes_the_ones_before_it(): + """Two values claiming to be primary are only counted once the write went through.""" + user = User( + user_name="bjensen", + title="Engineer", + emails=[Email(value="a@example.com"), Email(value="b@example.com")], + ) + patch = PatchOp[User]( + operations=[ + PatchOperation[User]( + op=PatchOperation.Op.replace_, path="title", value="Manager" + ), + PatchOperation[User]( + op=PatchOperation.Op.replace_, + path="emails", + value=[ + {"value": "a@example.com", "primary": True}, + {"value": "b@example.com", "primary": True}, + ], + ), + ] + ) + + with pytest.raises(Exception, match="Multiple values marked as primary"): + patch.patch(user) + + assert user.title == "Engineer" + assert [email.primary for email in user.emails] == [None, None] + + +def test_a_failing_patch_leaves_the_attributes_it_never_reached_alone(): + """Restoring the resource may not turn an unset attribute into a set one.""" + user = User(user_name="bjensen") + patch = PatchOp[User]( + operations=[ + PatchOperation[User]( + op=PatchOperation.Op.replace_, + path='emails[type eq "work"].value', + value="bjensen@example.com", + ), + ] + ) + + with pytest.raises(NoTargetException): + patch.patch(user) + + assert user.model_dump() == { + "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"], + "userName": "bjensen", + } + + +def test_the_caller_keeps_the_object_it_passed(): + """A server writes back the resource it read, so the patch may not swap it.""" + user = User(user_name="bjensen", emails=[Email(type="work", value="a@example.com")]) + patch = PatchOp[User]( + operations=[ + PatchOperation[User]( + op=PatchOperation.Op.replace_, + path='emails[type eq "work"].value', + value="b@example.com", + ) + ] + ) + same = user + + assert patch.patch(user) is True + assert same is user + assert user.emails[0].value == "b@example.com" + + +def test_a_patched_extension_survives_the_restoration(): + """An extension is held apart from the fields the resource declares.""" + user = User[EnterpriseUser]( + user_name="bjensen", + **{ + "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": EnterpriseUser( + department="Tour Operations" + ) + }, + ) + patch = PatchOp[User[EnterpriseUser]]( + operations=[ + PatchOperation[User[EnterpriseUser]]( + op=PatchOperation.Op.replace_, + path="urn:ietf:params:scim:schemas:extension:enterprise:2.0:User:department", + value="Sales", + ) + ] + ) + + assert patch.patch(user) is True + assert user[EnterpriseUser].department == "Sales" + + +def test_an_unknown_attribute_survives_the_restoration(): + """What a lenient policy set aside is held outside the fields, and is still reported.""" + user = User.model_validate( + { + "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"], + "userName": "bjensen", + "unknownAttr": "x", + }, + scim_policy=ScimPolicy(unknown=ScimPolicy.Unknown.ignore), + ) + patch = PatchOp[User]( + operations=[ + PatchOperation[User]( + op=PatchOperation.Op.replace_, path="title", value="Manager" + ) + ] + ) + + assert patch.patch(user) is True + assert user.unknown_attributes == {"unknownAttr": "x"}