Make patch operations atomic

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(

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"}