review fixes

docs/features/contacts.md

@@ -47,6 +47,6 @@ The following models support the assignment of contacts:
 
 ### Bulk Assignment
 
-Contacts can be assigned to or removed from many objects at once using the bulk edit form of any supported object type. When adding contacts, a role (and optionally a priority) must be selected to apply to each new assignment. Removing a contact deletes all of its assignments to the selected objects, regardless of role. Removals are processed before additions, so a contact's role can be replaced in a single operation.
+Contacts can be assigned to or removed from many objects at once using the bulk edit form of any supported object type. When adding contacts, a role (and optionally a priority) must be selected to apply to each new assignment. If a contact is already assigned to an object with the selected role, no duplicate is created; instead, the selected priority (if any) is applied to the existing assignment. Removing a contact deletes all of its assignments to the selected objects, regardless of role. Removals are processed before additions, so a contact's role can be replaced in a single operation.
 
 When importing contact assignments in bulk, contacts are matched by name by default. Where contact names are not unique, use the `contact.id` column header to reference contacts by their numeric ID instead.

netbox/netbox/forms/bulk_edit.py

@@ -48,7 +48,8 @@ class NetBoxModelBulkEditForm(ChangelogMessageMixin, CustomFieldsMixin, BulkEdit
     add_contacts = DynamicModelMultipleChoiceField(
         label=_('Add contacts'),
         queryset=Contact.objects.all(),
-        required=False
+        required=False,
+        help_text=_('Existing assignments with the selected role are updated to the selected priority, if any')
     )
     contact_role = DynamicModelChoiceField(
         label=_('Role'),

netbox/netbox/views/generic/bulk_views.py

@@ -922,11 +922,7 @@ class BulkEditView(GetReturnURLMixin, BaseMultiObjectView):
 
         # Removal precedes addition so that a contact's role can be replaced in a single operation
         if remove_contacts := form.cleaned_data.get('remove_contacts'):
-            to_delete = assignments.filter(contact__in=remove_contacts)
-            deletable = to_delete.restrict(self.request.user, 'delete')
-            if deletable.count() != to_delete.count():
-                raise PermissionsViolation
-            for assignment in to_delete:
+            for assignment in assignments.filter(contact__in=remove_contacts):
                 assignment.snapshot()
                 assignment._changelog_message = changelog_message
                 assignment.delete()
@@ -934,20 +930,59 @@ class BulkEditView(GetReturnURLMixin, BaseMultiObjectView):
         if add_contacts := form.cleaned_data.get('add_contacts'):
             role = form.cleaned_data.get('contact_role')
             priority = form.cleaned_data.get('contact_priority') or None
-            existing = set(assignments.filter(role=role).values_list('contact_id', flat=True))
-            created_pks = []
+            existing = {
+                a.contact_id: a for a in assignments.filter(role=role, contact__in=add_contacts)
+            }
+            cf_defaults = CustomField.objects.get_defaults_for_model(ContactAssignment)
             for contact in add_contacts:
-                if contact.pk in existing:
-                    continue
-                assignment = ContactAssignment(object=obj, contact=contact, role=role, priority=priority)
+                # Apply a specified priority to an existing assignment rather than duplicating it
+                if assignment := existing.get(contact.pk):
+                    if priority is None or assignment.priority == priority:
+                        continue
+                    assignment.snapshot()
+                    assignment.priority = priority
+                    pk_list = self._contact_assignments['change']
+                else:
+                    assignment = ContactAssignment(
+                        object=obj, contact=contact, role=role, priority=priority,
+                        custom_field_data=dict(cf_defaults),
+                    )
+                    pk_list = self._contact_assignments['add']
                 assignment._changelog_message = changelog_message
-                assignment.full_clean()
+                try:
+                    assignment.full_clean()
+                except ValidationError as e:
+                    raise ValidationError(
+                        _("Unable to assign contact {contact} to {object}: {error}").format(
+                            contact=contact, object=obj, error='; '.join(e.messages)
+                        )
+                    )
                 assignment.save()
-                created_pks.append(assignment.pk)
-            permitted = ContactAssignment.objects.restrict(self.request.user, 'add').filter(pk__in=created_pks)
-            if permitted.count() != len(created_pks):
+                pk_list.append(assignment.pk)
+
+    def _check_contact_removal_permission(self, form):
+        """
+        Verify that the user may delete every contact assignment targeted for removal.
+        """
+        if remove_contacts := form.cleaned_data.get('remove_contacts'):
+            to_delete = ContactAssignment.objects.filter(
+                object_type=ObjectType.objects.get_for_model(self.queryset.model),
+                object_id__in=self.queryset.filter(pk__in=form.cleaned_data['pk']).values('pk'),
+                contact__in=remove_contacts,
+            )
+            if to_delete.restrict(self.request.user, 'delete').count() != to_delete.count():
                 raise PermissionsViolation
 
+    def _check_contact_assignment_permissions(self):
+        """
+        Verify that the user was permitted to create or modify every affected contact assignment.
+        """
+        for action, pk_list in self._contact_assignments.items():
+            if pk_list:
+                permitted = ContactAssignment.objects.restrict(self.request.user, action).filter(pk__in=pk_list)
+                if permitted.count() != len(pk_list):
+                    raise PermissionsViolation
+
     def _update_objects(self, form, request):
         custom_fields = getattr(form, 'custom_fields', {})
         standard_fields = [
@@ -973,6 +1008,10 @@ class BulkEditView(GetReturnURLMixin, BaseMultiObjectView):
                 # This form field is used to modify a field rather than set its value directly
                 model_fields[name] = None
 
+        # Contact assignment permissions are checked once for all objects rather than per object
+        self._contact_assignments = {'add': [], 'change': []}
+        self._check_contact_removal_permission(form)
+
         for obj in self.queryset.filter(pk__in=form.cleaned_data['pk']):
 
             # Take a snapshot of change-logged models
@@ -1034,6 +1073,8 @@ class BulkEditView(GetReturnURLMixin, BaseMultiObjectView):
             if is_background_request(request):
                 request.job.logger.info(f"Updated {obj}")
 
+        self._check_contact_assignment_permissions()
+
         # Rebuild the tree for MPTT plugin models (no-op for ltree; its triggers keep
         # the tree current). TODO: Remove in v5.0 (see _rebuild_mptt_tree).
         _rebuild_mptt_tree(self.queryset.model)

netbox/tenancy/tests/test_views.py

@@ -1,7 +1,10 @@
 from django.contrib.contenttypes.models import ContentType
 from django.urls import reverse
 
+from core.models import ObjectType
 from dcim.models import Site
+from extras.choices import CustomFieldTypeChoices
+from extras.models import CustomField
 from tenancy.choices import ContactPriorityChoices
 from tenancy.forms import ContactBulkEditForm
 from tenancy.models import *
@@ -136,7 +139,10 @@ class TenantBulkEditContactsTestCase(TestCase):
             'tenancy.add_contactassignment',
         )
         # An existing assignment should not be duplicated
-        ContactAssignment.objects.create(object=self.tenants[0], contact=self.contacts[0], role=self.roles[0])
+        ContactAssignment.objects.create(
+            object=self.tenants[0], contact=self.contacts[0], role=self.roles[0],
+            priority=ContactPriorityChoices.PRIORITY_PRIMARY,
+        )
 
         response = self._post(
             add_contacts=[c.pk for c in self.contacts],
@@ -151,7 +157,92 @@ class TenantBulkEditContactsTestCase(TestCase):
             self.assertEqual(
                 set(assignments.values_list('contact', flat=True)), {c.pk for c in self.contacts}
             )
-        self.assertEqual(ContactAssignment.objects.filter(priority=ContactPriorityChoices.PRIORITY_PRIMARY).count(), 3)
+        self.assertEqual(ContactAssignment.objects.filter(priority=ContactPriorityChoices.PRIORITY_PRIMARY).count(), 4)
+
+    def test_add_contacts_without_priority(self):
+        self.add_permissions(
+            'tenancy.view_tenant', 'tenancy.change_tenant', 'tenancy.view_contact', 'tenancy.view_contactrole',
+            'tenancy.add_contactassignment',
+        )
+        response = self._post(add_contacts=[self.contacts[0].pk], contact_role=self.roles[0].pk)
+        self.assertHttpStatus(response, 302)
+        self.assertEqual(ContactAssignment.objects.filter(contact=self.contacts[0], priority__isnull=True).count(), 2)
+
+    def test_add_contacts_updates_existing_priority(self):
+        self.add_permissions(
+            'tenancy.view_tenant', 'tenancy.change_tenant', 'tenancy.view_contact', 'tenancy.view_contactrole',
+            'tenancy.add_contactassignment', 'tenancy.change_contactassignment',
+        )
+        assignment = ContactAssignment.objects.create(
+            object=self.tenants[0], contact=self.contacts[0], role=self.roles[0]
+        )
+
+        response = self._post(
+            add_contacts=[self.contacts[0].pk],
+            contact_role=self.roles[0].pk,
+            contact_priority=ContactPriorityChoices.PRIORITY_PRIMARY,
+        )
+        self.assertHttpStatus(response, 302)
+        assignment.refresh_from_db()
+        self.assertEqual(assignment.priority, ContactPriorityChoices.PRIORITY_PRIMARY)
+        self.assertEqual(ContactAssignment.objects.count(), 2)
+
+    def test_add_contacts_updates_existing_priority_without_permission(self):
+        self.add_permissions(
+            'tenancy.view_tenant', 'tenancy.change_tenant', 'tenancy.view_contact', 'tenancy.view_contactrole',
+            'tenancy.add_contactassignment',
+        )
+        assignment = ContactAssignment.objects.create(
+            object=self.tenants[0], contact=self.contacts[0], role=self.roles[0]
+        )
+
+        response = self._post(
+            add_contacts=[self.contacts[0].pk],
+            contact_role=self.roles[0].pk,
+            contact_priority=ContactPriorityChoices.PRIORITY_PRIMARY,
+        )
+        self.assertHttpStatus(response, 200)
+        assignment.refresh_from_db()
+        self.assertIsNone(assignment.priority)
+        self.assertEqual(ContactAssignment.objects.count(), 1)
+
+    def test_add_contacts_required_custom_field(self):
+        self.add_permissions(
+            'tenancy.view_tenant', 'tenancy.change_tenant', 'tenancy.view_contact', 'tenancy.view_contactrole',
+            'tenancy.add_contactassignment',
+        )
+        cf = CustomField.objects.create(name='cf1', type=CustomFieldTypeChoices.TYPE_TEXT, required=True)
+        cf.object_types.set([ObjectType.objects.get_for_model(ContactAssignment)])
+
+        # A required custom field without a default cannot be satisfied
+        response = self._post(add_contacts=[self.contacts[0].pk], contact_role=self.roles[0].pk)
+        self.assertHttpStatus(response, 200)
+        self.assertFalse(ContactAssignment.objects.exists())
+
+        # The default value is applied to new assignments
+        cf.default = 'foo'
+        cf.save()
+        response = self._post(add_contacts=[self.contacts[0].pk], contact_role=self.roles[0].pk)
+        self.assertHttpStatus(response, 302)
+        self.assertEqual(ContactAssignment.objects.filter(custom_field_data__cf1='foo').count(), 2)
+
+    def test_replace_contact_role(self):
+        self.add_permissions(
+            'tenancy.view_tenant', 'tenancy.change_tenant', 'tenancy.view_contact', 'tenancy.view_contactrole',
+            'tenancy.add_contactassignment', 'tenancy.delete_contactassignment',
+        )
+        for tenant in self.tenants:
+            ContactAssignment.objects.create(object=tenant, contact=self.contacts[0], role=self.roles[0])
+
+        # Removing and re-adding a contact in one edit replaces its role
+        response = self._post(
+            add_contacts=[self.contacts[0].pk],
+            contact_role=self.roles[1].pk,
+            remove_contacts=[self.contacts[0].pk],
+        )
+        self.assertHttpStatus(response, 302)
+        self.assertEqual(ContactAssignment.objects.filter(contact=self.contacts[0], role=self.roles[1]).count(), 2)
+        self.assertFalse(ContactAssignment.objects.filter(role=self.roles[0]).exists())
 
     def test_add_contacts_requires_role(self):
         self.add_permissions(