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

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. 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,29 @@ class NetBoxModelBulkEditForm(ChangelogMessageMixin, CustomFieldsMixin, BulkEdit
         queryset=Tag.objects.all(),
         required=False
     )
+    add_contacts = DynamicModelMultipleChoiceField(
+        label=_('Add contacts'),
+        queryset=Contact.objects.all(),
+        required=False
+    )
+    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 +79,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,43 @@ 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'):
+            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:
+                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 = set(assignments.filter(role=role).values_list('contact_id', flat=True))
+            created_pks = []
+            for contact in add_contacts:
+                if contact.pk in existing:
+                    continue
+                assignment = ContactAssignment(object=obj, contact=contact, role=role, priority=priority)
+                assignment._changelog_message = changelog_message
+                assignment.full_clean()
+                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):
+                raise PermissionsViolation
+
     def _update_objects(self, form, request):
         custom_fields = getattr(form, 'custom_fields', {})
         standard_fields = [

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

@@ -3,8 +3,9 @@ from django.urls import reverse
 
 from dcim.models import Site
 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 +103,104 @@ 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])
+
+        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(), 3)
+
+    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