Closes #19515: Enable bulk assignment & removal of contacts via bulk edit (#23672)

* Closes #19515: Enable bulk assignment & removal of contacts via bulk edit

* review fixes

docs/features/contacts.md

@@ -44,3 +44,9 @@ The following models support the assignment of contacts:
 * virtualization.Cluster
 * virtualization.ClusterGroup
 * virtualization.VirtualMachine
+
+### 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. 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

@@ -4,8 +4,11 @@ from django.utils.translation import gettext_lazy as _
 from core.models import ObjectType
 from extras.choices import *
 from extras.models import Tag
-from utilities.forms import BulkEditForm
-from utilities.forms.fields import CommentField, DynamicModelMultipleChoiceField
+from netbox.models.features import ContactsMixin
+from tenancy.choices import ContactPriorityChoices
+from tenancy.models import Contact, ContactRole
+from utilities.forms import BulkEditForm, add_blank_choice
+from utilities.forms.fields import CommentField, DynamicModelChoiceField, DynamicModelMultipleChoiceField
 
 from .mixins import ChangelogMessageMixin, CustomFieldsMixin, OwnerMixin
 
@@ -20,7 +23,7 @@ __all__ = (
 class NetBoxModelBulkEditForm(ChangelogMessageMixin, CustomFieldsMixin, BulkEditForm):
     """
     Base form for modifying multiple NetBox objects (of the same type) in bulk via the UI. Adds support for custom
-    fields and adding/removing tags.
+    fields, adding/removing tags, and adding/removing contact assignments (for models which support contacts).
 
     Attributes:
         fieldsets: An iterable of two-tuples which define a heading and field set to display per section of
@@ -42,6 +45,30 @@ class NetBoxModelBulkEditForm(ChangelogMessageMixin, CustomFieldsMixin, BulkEdit
         queryset=Tag.objects.all(),
         required=False
     )
+    add_contacts = DynamicModelMultipleChoiceField(
+        label=_('Add contacts'),
+        queryset=Contact.objects.all(),
+        required=False,
+        help_text=_('Existing assignments with the selected role are updated to the selected priority, if any')
+    )
+    contact_role = DynamicModelChoiceField(
+        label=_('Role'),
+        queryset=ContactRole.objects.all(),
+        required=False,
+        help_text=_('Role to assign to added contacts')
+    )
+    contact_priority = forms.ChoiceField(
+        label=_('Priority'),
+        choices=add_blank_choice(ContactPriorityChoices),
+        required=False,
+        help_text=_('Priority to assign to added contacts')
+    )
+    remove_contacts = DynamicModelMultipleChoiceField(
+        label=_('Remove contacts'),
+        queryset=Contact.objects.all(),
+        required=False,
+        help_text=_('Removes all assignments of these contacts, regardless of role')
+    )
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -53,8 +80,23 @@ class NetBoxModelBulkEditForm(ChangelogMessageMixin, CustomFieldsMixin, BulkEdit
         self.fields['add_tags'].widget.add_query_param('for_object_type_id', object_type.pk)
         self.fields['remove_tags'].widget.add_query_param('for_object_type_id', object_type.pk)
 
+        # Contact fields apply only to models which support contact assignment
+        if not issubclass(self.model, ContactsMixin):
+            for name in ('add_contacts', 'contact_role', 'contact_priority', 'remove_contacts'):
+                del self.fields[name]
+
         self._extend_nullable_fields()
 
+    def clean(self):
+        super().clean()
+
+        if self.cleaned_data.get('add_contacts') and not self.cleaned_data.get('contact_role'):
+            raise forms.ValidationError({
+                'contact_role': _('A role must be specified when adding contacts.')
+            })
+
+        return self.cleaned_data
+
     def _get_form_field(self, customfield):
         return customfield.to_form_field(set_initial=False, enforce_required=False)
 

netbox/netbox/views/generic/bulk_views.py

@@ -33,6 +33,7 @@ from netbox.exporters import get_exporter
 from netbox.forms.bulk_rename import NetBoxModelBulkRenameForm
 from netbox.models.features import ChangeLoggingMixin
 from netbox.object_actions import AddObject, BulkDelete, BulkEdit, BulkExport, BulkImport, BulkRename
+from tenancy.models import ContactAssignment
 from utilities.error_handlers import handle_protectederror
 from utilities.exceptions import AbortRequest, AbortTransaction, PermissionsViolation
 from utilities.forms import BulkDeleteForm, BulkRenameForm, restrict_form_fields
@@ -910,6 +911,78 @@ class BulkEditView(GetReturnURLMixin, BaseMultiObjectView):
         if form.cleaned_data.get('remove_tags', None):
             obj.tags.remove(*form.cleaned_data['remove_tags'])
 
+        # Remove/add contact assignments
+        if form.cleaned_data.get('remove_contacts') or form.cleaned_data.get('add_contacts'):
+            self._update_contact_assignments(form, obj)
+
+    def _update_contact_assignments(self, form, obj):
+        object_type = ObjectType.objects.get_for_model(obj)
+        assignments = ContactAssignment.objects.filter(object_type=object_type, object_id=obj.pk)
+        changelog_message = form.cleaned_data.get('changelog_message')
+
+        # 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'):
+            for assignment in assignments.filter(contact__in=remove_contacts):
+                assignment.snapshot()
+                assignment._changelog_message = changelog_message
+                assignment.delete()
+
+        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 = {
+                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:
+                # 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
+                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()
+                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 = [
@@ -935,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
@@ -996,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/templates/generic/bulk_edit.html

@@ -82,6 +82,19 @@ Context:
             </div>
           {% endif %}
 
+          {# Render contact add/remove fields #}
+          {% if form.add_contacts and form.remove_contacts %}
+            <div class="field-group mb-5">
+              <div class="row">
+                <h2 class="col-9 offset-3">{% trans "Contacts" %}</h2>
+              </div>
+              {% render_field form.add_contacts %}
+              {% render_field form.contact_role %}
+              {% render_field form.contact_priority %}
+              {% render_field form.remove_contacts %}
+            </div>
+          {% endif %}
+
           {# Render custom fields #}
           {% if form.custom_fields %}
             <div class="field-group mb-5">

netbox/tenancy/tests/test_views.py

@@ -1,10 +1,14 @@
 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 *
-from utilities.testing import ViewTestCases, create_tags
+from utilities.testing import TestCase, ViewTestCases, create_tags
 
 
 class TenantGroupTestCase(ViewTestCases.OrganizationalObjectViewTestCase):
@@ -102,6 +106,192 @@ class TenantTestCase(ViewTestCases.PrimaryObjectViewTestCase):
         }
 
 
+class TenantBulkEditContactsTestCase(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.tenants = (
+            Tenant(name='Tenant 1', slug='tenant-1'),
+            Tenant(name='Tenant 2', slug='tenant-2'),
+        )
+        Tenant.objects.bulk_create(cls.tenants)
+        cls.contacts = (
+            Contact(name='Contact 1'),
+            Contact(name='Contact 2'),
+        )
+        Contact.objects.bulk_create(cls.contacts)
+        cls.roles = (
+            ContactRole(name='Role 1', slug='role-1'),
+            ContactRole(name='Role 2', slug='role-2'),
+        )
+        ContactRole.objects.bulk_create(cls.roles)
+
+    def _post(self, **data):
+        return self.client.post(reverse('tenancy:tenant_bulk_edit'), {
+            'pk': [t.pk for t in self.tenants],
+            '_apply': True,
+            **data,
+        })
+
+    def test_add_contacts(self):
+        self.add_permissions(
+            'tenancy.view_tenant', 'tenancy.change_tenant', 'tenancy.view_contact', 'tenancy.view_contactrole',
+            '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],
+            priority=ContactPriorityChoices.PRIORITY_PRIMARY,
+        )
+
+        response = self._post(
+            add_contacts=[c.pk for c in self.contacts],
+            contact_role=self.roles[0].pk,
+            contact_priority=ContactPriorityChoices.PRIORITY_PRIMARY,
+        )
+        self.assertHttpStatus(response, 302)
+        for tenant in self.tenants:
+            assignments = ContactAssignment.objects.filter(
+                object_type=ContentType.objects.get_for_model(Tenant), object_id=tenant.pk, role=self.roles[0]
+            )
+            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(), 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(
+            'tenancy.view_tenant', 'tenancy.change_tenant', 'tenancy.view_contact',
+            'tenancy.add_contactassignment',
+        )
+        response = self._post(add_contacts=[self.contacts[0].pk])
+        self.assertHttpStatus(response, 200)
+        self.assertFalse(ContactAssignment.objects.exists())
+
+    def test_add_contacts_without_permission(self):
+        self.add_permissions(
+            'tenancy.view_tenant', 'tenancy.change_tenant', 'tenancy.view_contact', 'tenancy.view_contactrole',
+        )
+        response = self._post(add_contacts=[self.contacts[0].pk], contact_role=self.roles[0].pk)
+        self.assertHttpStatus(response, 200)
+        self.assertFalse(ContactAssignment.objects.exists())
+
+    def test_remove_contacts(self):
+        self.add_permissions(
+            'tenancy.view_tenant', 'tenancy.change_tenant', 'tenancy.view_contact',
+            'tenancy.delete_contactassignment',
+        )
+        for tenant in self.tenants:
+            for role in self.roles:
+                ContactAssignment.objects.create(object=tenant, contact=self.contacts[0], role=role)
+            ContactAssignment.objects.create(object=tenant, contact=self.contacts[1], role=self.roles[0])
+
+        response = self._post(remove_contacts=[self.contacts[0].pk])
+        self.assertHttpStatus(response, 302)
+        self.assertFalse(ContactAssignment.objects.filter(contact=self.contacts[0]).exists())
+        self.assertEqual(ContactAssignment.objects.filter(contact=self.contacts[1]).count(), 2)
+
+    def test_remove_contacts_without_permission(self):
+        self.add_permissions(
+            'tenancy.view_tenant', 'tenancy.change_tenant', 'tenancy.view_contact',
+        )
+        ContactAssignment.objects.create(object=self.tenants[0], contact=self.contacts[0], role=self.roles[0])
+
+        response = self._post(remove_contacts=[self.contacts[0].pk])
+        self.assertHttpStatus(response, 200)
+        self.assertEqual(ContactAssignment.objects.count(), 1)
+
+    def test_contact_fields_omitted_for_unsupported_models(self):
+        form = ContactBulkEditForm()
+        self.assertNotIn('add_contacts', form.fields)
+        self.assertNotIn('remove_contacts', form.fields)
+
+
 class ContactGroupTestCase(ViewTestCases.OrganizationalObjectViewTestCase):
     model = ContactGroup