Merge pull request #120 from python-scim/10-one-primary

Primary attribute auto-exclusion

Author: azmeuk

doc/changelog.rst

@@ -11,6 +11,7 @@ Added
 - Validation that extension schemas are known during SCIM context validation.
 - Introduce SCIM exceptions hierarchy (:class:`~scim2_models.SCIMException` and subclasses) corresponding to RFC 7644 error types. :issue:`103`
 - :meth:`Error.from_validation_error <scim2_models.Error.from_validation_error>` to convert Pydantic :class:`~pydantic.ValidationError` to SCIM :class:`~scim2_models.Error`.
+- :meth:`PatchOp.patch <scim2_models.PatchOp.patch>` auto-excludes other ``primary`` values when setting one to ``True``. :issue:`116`
 
 Changed
 ^^^^^^^

scim2_models/attributes.py

@@ -34,7 +34,11 @@ class MultiValuedComplexAttribute(ComplexAttribute):
 
     primary: bool | None = None
     """A Boolean value indicating the 'primary' or preferred attribute value
-    for this attribute."""
+    for this attribute.
+
+    Per :rfc:`RFC 7643 §2.4 <7643#section-2.4>`, the primary attribute value
+    ``True`` MUST appear no more than once in a multi-valued attribute list.
+    """
 
     display: Annotated[str | None, Mutability.immutable] = None
     """A human-readable name, primarily used for display purposes."""

scim2_models/messages/patch_op.py

@@ -5,6 +5,7 @@
 from typing import Generic
 from typing import TypeVar
 
+from pydantic import BaseModel as PydanticBaseModel
 from pydantic import Field
 from pydantic import ValidationInfo
 from pydantic import field_validator
@@ -257,10 +258,14 @@ def patch(self, resource: ResourceT) -> bool:
         "add", "replace", and "remove". If any operation modifies the resource, the method
         returns True; otherwise, False.
 
+        Per :rfc:`RFC 7644 §3.5.2 <7644#section-3.5.2>`, when an operation sets a value's
+        ``primary`` sub-attribute to ``True``, any other values in the same multi-valued
+        attribute will have their ``primary`` set to ``False`` automatically.
+
         :param resource: The SCIM resource to patch. This object is modified in-place.
-        :type resource: T
         :return: True if the resource was modified by any operation, False otherwise.
-        :raises SCIMException: If an operation is invalid (e.g., invalid path, forbidden mutation).
+        :raises InvalidValueException: If multiple values are marked as primary in a single
+            operation, or if multiple primary values already exist before the patch.
         """
         if not self.operations:
             return False
@@ -291,13 +296,102 @@ def _apply_add_replace(
         self, resource: Resource[Any], operation: PatchOperation[ResourceT]
     ) -> bool:
         """Apply an add or replace operation."""
+        before_state = self._capture_primary_state(resource)
+
         path = operation.path if operation.path is not None else Path("")
-        return path.set(
+        modified = path.set(
             resource,  # type: ignore[arg-type]
             operation.value,
             is_add=operation.op == PatchOperation.Op.add,
         )
 
+        if modified:
+            self._normalize_primary_after_patch(resource, before_state)
+
+        return modified
+
+    def _capture_primary_state(self, resource: Resource[Any]) -> dict[str, set[int]]:
+        """Capture indices of elements with primary=True for each multi-valued attribute."""
+        state: dict[str, set[int]] = {}
+        for field_name in type(resource).model_fields:
+            if not resource.get_field_multiplicity(field_name):
+                continue
+
+            field_value = getattr(resource, field_name, None)
+            if not field_value:
+                continue
+
+            element_type = resource.get_field_root_type(field_name)
+            if (
+                not element_type
+                or not isclass(element_type)
+                or not issubclass(element_type, PydanticBaseModel)
+                or "primary" not in element_type.model_fields
+            ):
+                continue
+
+            primary_indices = {
+                i
+                for i, item in enumerate(field_value)
+                if getattr(item, "primary", None) is True
+            }
+            state[field_name] = primary_indices
+
+        return state
+
+    def _normalize_primary_after_patch(
+        self, resource: Resource[Any], before_state: dict[str, set[int]]
+    ) -> None:
+        """Normalize primary attributes after a patch operation.
+
+        Per :rfc:`RFC 7644 §3.5.2 <7644#section-3.5.2>`: a PATCH operation that
+        sets a value's "primary" sub-attribute to "true" SHALL cause the server
+        to automatically set "primary" to "false" for any other values.
+        """
+        for field_name in type(resource).model_fields:
+            if not resource.get_field_multiplicity(field_name):
+                continue
+
+            field_value = getattr(resource, field_name, None)
+            if not field_value:
+                continue
+
+            element_type = resource.get_field_root_type(field_name)
+            if (
+                not element_type
+                or not isclass(element_type)
+                or not issubclass(element_type, PydanticBaseModel)
+                or "primary" not in element_type.model_fields
+            ):
+                continue
+
+            current_primary_indices = {
+                i
+                for i, item in enumerate(field_value)
+                if getattr(item, "primary", None) is True
+            }
+
+            if len(current_primary_indices) <= 1:
+                continue
+
+            before_primaries = before_state.get(field_name, set())
+            new_primaries = current_primary_indices - before_primaries
+
+            if len(new_primaries) > 1:
+                raise InvalidValueException(
+                    detail=f"Multiple values marked as primary in field '{field_name}'"
+                )
+
+            if not new_primaries:
+                raise InvalidValueException(
+                    detail=f"Multiple primary values already exist in field '{field_name}'"
+                )
+
+            keep_index = next(iter(new_primaries))
+            for i in current_primary_indices:
+                if i != keep_index:
+                    field_value[i].primary = False
+
     def _apply_remove(
         self, resource: Resource[Any], operation: PatchOperation[ResourceT]
     ) -> bool:

tests/test_patch_op_replace.py

@@ -219,3 +219,147 @@ class Dummy(Resource):
                 )
             ]
         )
+
+
+def test_primary_auto_exclusion_on_add():
+    """Test that adding an element with primary=true auto-excludes other primary values.
+
+    :rfc:`RFC 7644 §3.5.2 <7644#section-3.5.2>`: "a PATCH operation that sets a
+    value's 'primary' sub-attribute to 'true' SHALL cause the server to
+    automatically set 'primary' to 'false' for any other values in the array."
+    """
+    from scim2_models import Email
+
+    user = User(
+        emails=[
+            Email(value="existing@example.com", primary=True),
+        ]
+    )
+
+    patch = PatchOp[User](
+        operations=[
+            PatchOperation[User](
+                op=PatchOperation.Op.add,
+                path="emails",
+                value={"value": "new@example.com", "primary": True},
+            )
+        ]
+    )
+
+    result = patch.patch(user)
+
+    assert result is True
+    assert user.emails[0].primary is False
+    assert user.emails[1].primary is True
+
+
+def test_primary_auto_exclusion_on_replace_list():
+    """Test that replacing a list with a new primary auto-excludes the old one."""
+    from scim2_models import Email
+
+    user = User(
+        emails=[
+            Email(value="old@example.com", primary=True),
+        ]
+    )
+
+    patch = PatchOp[User](
+        operations=[
+            PatchOperation[User](
+                op=PatchOperation.Op.replace_,
+                path="emails",
+                value=[
+                    {"value": "old@example.com", "primary": False},
+                    {"value": "new@example.com", "primary": True},
+                ],
+            )
+        ]
+    )
+
+    result = patch.patch(user)
+
+    assert result is True
+    assert user.emails[0].primary is False
+    assert user.emails[1].primary is True
+
+
+def test_primary_no_change_when_single_primary():
+    """Test that no change occurs when there's only one primary after patch."""
+    from scim2_models import Email
+
+    user = User(
+        emails=[
+            Email(value="a@example.com", primary=True),
+            Email(value="b@example.com", primary=False),
+        ]
+    )
+
+    patch = PatchOp[User](
+        operations=[
+            PatchOperation[User](
+                op=PatchOperation.Op.add,
+                path="emails",
+                value={"value": "c@example.com", "primary": False},
+            )
+        ]
+    )
+
+    result = patch.patch(user)
+
+    assert result is True
+    assert user.emails[0].primary is True
+    assert user.emails[1].primary is False
+    assert user.emails[2].primary is False
+
+
+def test_primary_auto_exclusion_rejects_multiple_new_primaries():
+    """Test that setting multiple new primaries in one operation raises an error."""
+    from scim2_models import Email
+
+    user = User(
+        emails=[
+            Email(value="a@example.com", primary=False),
+            Email(value="b@example.com", primary=False),
+        ]
+    )
+
+    patch = PatchOp[User](
+        operations=[
+            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)
+
+
+def test_primary_auto_exclusion_rejects_preexisting_multiple_primaries():
+    """Test that patching data with preexisting multiple primaries raises an error."""
+    from scim2_models import Email
+
+    user = User(
+        emails=[
+            Email(value="a@example.com", primary=True),
+            Email(value="b@example.com", primary=True),
+        ]
+    )
+
+    patch = PatchOp[User](
+        operations=[
+            PatchOperation[User](
+                op=PatchOperation.Op.add,
+                path="emails",
+                value={"value": "c@example.com", "primary": False},
+            )
+        ]
+    )
+
+    with pytest.raises(Exception, match="Multiple primary values already exist"):
+        patch.patch(user)