Merge pull request #22901 from netbox-community/20054-cleanup

#20054: Pre-release QA

docs/integrations/rest-api.md

@@ -592,6 +592,9 @@ http://netbox/api/dcim/sites/ \
 ]
 ```
 
+!!! note
+    The bulk creation of objects is an all-or-none operation, meaning that if NetBox fails to successfully create any of the specified objects (e.g. due to a validation error), the entire operation will be aborted and none of the objects will be created.
+
 ### Updating an Object
 
 To modify an object which has already been created, make a `PATCH` request to the model's _detail_ endpoint specifying its unique numeric ID. Include any data which you wish to update on the object. As with object creation, the `Authorization` and `Content-Type` headers must also be specified.

netbox/core/api/schema.py

@@ -2,6 +2,7 @@ import re
 import typing
 from collections import OrderedDict
 
+from django.utils.translation import gettext_lazy as _
 from drf_spectacular.contrib.django_filters import DjangoFilterExtension
 from drf_spectacular.extensions import OpenApiSerializerExtension, OpenApiSerializerFieldExtension, _SchemaType
 from drf_spectacular.openapi import AutoSchema
@@ -14,10 +15,10 @@ from drf_spectacular.plumbing import (
     get_doc,
 )
 from drf_spectacular.types import OpenApiTypes
-from drf_spectacular.utils import Direction, OpenApiParameter
+from drf_spectacular.utils import Direction, OpenApiParameter, OpenApiResponse
 
 from netbox.api.fields import ChoiceField
-from netbox.api.serializers import WritableNestedSerializer
+from netbox.api.serializers import BulkOperationErrorSerializer, WritableNestedSerializer
 from netbox.api.viewsets import NetBoxModelViewSet
 
 # see netbox.api.routers.NetBoxRouter
@@ -182,6 +183,106 @@ class NetBoxAutoSchema(AutoSchema):
 
         return response_serializers
 
+    def _get_bulk_error_responses(self, direction) -> typing.Any:
+        """
+        Return the error responses of the current bulk write action, keyed by status code, or an
+        empty dict if this action is not a bulk write.
+
+        A failed bulk write returns a structured body correlating each failure with the object (or,
+        where no object could be identified, the request position) responsible for it. This is a
+        documented part of the API contract, but drf-spectacular cannot infer it: responses are
+        derived from the request/response serializer alone, which describes only the success case.
+        """
+        action = getattr(self.view, 'action', None)
+
+        if action in ('bulk_update', 'bulk_partial_update'):
+            return {
+                '400': OpenApiResponse(
+                    response=BulkOperationErrorSerializer,
+                    description=_(
+                        "One or more of the objects specified could not be updated. No objects were "
+                        "modified: a bulk update is an all-or-none operation."
+                    ),
+                ),
+                '403': OpenApiResponse(
+                    response=BulkOperationErrorSerializer,
+                    description=_(
+                        "The requesting user is not permitted to apply one or more of the "
+                        "modifications specified. No objects were modified."
+                    ),
+                ),
+            }
+
+        if action == 'bulk_destroy':
+            return {
+                '400': OpenApiResponse(
+                    response=BulkOperationErrorSerializer,
+                    description=_(
+                        "The request was malformed, one or more of the objects specified could not "
+                        "be found, or the deletion of one of them was prevented by a protection "
+                        "rule. No objects were deleted."
+                    ),
+                ),
+                '403': OpenApiResponse(
+                    response=BulkOperationErrorSerializer,
+                    description=_(
+                        "The requesting user is not permitted to delete one or more of the objects "
+                        "specified. No objects were deleted."
+                    ),
+                ),
+                '409': OpenApiResponse(
+                    response=BulkOperationErrorSerializer,
+                    description=_(
+                        "One or more of the objects specified could not be deleted, because a "
+                        "dependent object prevents it. No objects were deleted: a bulk deletion is "
+                        "an all-or-none operation."
+                    ),
+                ),
+            }
+
+        if action == 'create' and viewset_handles_bulk_create(self.view):
+            # A POST to a list endpoint accepts either a single object or a list of them (see
+            # _get_request_for_media_type()), so its error body takes one of two shapes
+            # accordingly: field-keyed errors for a single object, or the bulk envelope for a list.
+            component = self.resolve_serializer(BulkOperationErrorSerializer, direction)
+            return {
+                '400': OpenApiResponse(
+                    response={
+                        'oneOf': [
+                            build_basic_type(OpenApiTypes.OBJECT),
+                            component.ref if component else build_basic_type(OpenApiTypes.OBJECT),
+                        ],
+                    },
+                    description=_(
+                        "The object could not be created. Where a list was submitted, no objects "
+                        "were created: a bulk creation is an all-or-none operation."
+                    ),
+                ),
+                # A 403 always carries a `detail`, and BulkOperationError's `errors` is optional, so
+                # the one component covers both the single-object and the bulk shape here.
+                '403': OpenApiResponse(
+                    response=BulkOperationErrorSerializer,
+                    description=_(
+                        "The requesting user is not permitted to create one or more of the objects "
+                        "specified. No objects were created."
+                    ),
+                ),
+            }
+
+        return {}
+
+    def _get_response_bodies(self, direction='response') -> typing.Any:
+        responses = super()._get_response_bodies(direction=direction)
+
+        # Document the error responses of the bulk write actions, which cannot be inferred (see
+        # _get_bulk_error_responses). A status code already present -- for instance one declared
+        # via @extend_schema on a custom action -- is left as it is.
+        for code, response in self._get_bulk_error_responses(direction).items():
+            if code not in responses:
+                responses[code] = self._get_response_for_code(response, code, direction=direction)
+
+        return responses
+
     def _get_request_for_media_type(self, serializer, direction='request'):
         """
         Override to generate oneOf schema for serializers that support both

netbox/core/tests/test_openapi_schema.py

@@ -107,3 +107,114 @@ class OpenAPISchemaTestCase(TestCase):
         self.assertNotIn('oneOf', request_schema, "DELETE should NOT have oneOf")
         self.assertEqual(request_schema['type'], 'array', "DELETE should require array")
         self.assertIn('items', request_schema, "DELETE array should have items")
+
+    def _get_response_schema(self, path, method, code):
+        """Return the JSON response schema documented for the given operation and status code."""
+        responses = self.schema['paths'][path][method]['responses']
+        self.assertIn(code, responses, f"{method.upper()} {path} should document a {code} response")
+        return responses[code]['content']['application/json']['schema']
+
+    def test_bulk_error_component_is_defined(self):
+        """
+        The structured error body returned by a failed bulk operation should be a named component,
+        so that generated clients have a type for it.
+
+        Refs: #20054
+        """
+        components = self.schema['components']['schemas']
+
+        self.assertIn('BulkOperationError', components)
+        envelope = components['BulkOperationError']
+        self.assertEqual(sorted(envelope['properties']), ['detail', 'errors'])
+        # `errors` is absent where the request could not be attributed to individual entries
+        self.assertEqual(envelope['required'], ['detail'])
+        self.assertEqual(
+            envelope['properties']['errors']['items']['$ref'],
+            '#/components/schemas/BulkOperationEntryError',
+        )
+
+        self.assertIn('BulkOperationEntryError', components)
+        entry = components['BulkOperationEntryError']
+        # An entry is correlated by `id` or by `index`, so neither is required; `errors` always is
+        self.assertEqual(sorted(entry['properties']), ['errors', 'id', 'index'])
+        self.assertEqual(entry['required'], ['errors'])
+
+    def test_bulk_update_documents_error_response(self):
+        """
+        Bulk update operations should document the structured 400 response.
+
+        Refs: #20054
+        """
+        ref = {'$ref': '#/components/schemas/BulkOperationError'}
+
+        for path in ('/api/dcim/sites/', '/api/ipam/prefixes/', '/api/users/users/'):
+            for method in ('put', 'patch'):
+                with self.subTest(path=path, method=method):
+                    self.assertEqual(self._get_response_schema(path, method, '400'), ref)
+
+    def test_bulk_delete_documents_error_responses(self):
+        """
+        Bulk delete operations should document the 400 (unresolvable request or protection rule), the
+        403 (not permitted) and the 409 (dependent object) responses.
+
+        Refs: #20054
+        """
+        ref = {'$ref': '#/components/schemas/BulkOperationError'}
+
+        for path in ('/api/dcim/sites/', '/api/ipam/prefixes/', '/api/users/users/'):
+            with self.subTest(path=path):
+                self.assertEqual(self._get_response_schema(path, 'delete', '400'), ref)
+                self.assertEqual(self._get_response_schema(path, 'delete', '403'), ref)
+                self.assertEqual(self._get_response_schema(path, 'delete', '409'), ref)
+
+    def test_bulk_write_operations_document_forbidden_response(self):
+        """
+        Every bulk write should document the 403 returned when an object-level permission refuses one
+        of the objects specified.
+
+        Refs: #20054
+        """
+        ref = {'$ref': '#/components/schemas/BulkOperationError'}
+
+        for path in ('/api/dcim/sites/', '/api/ipam/prefixes/', '/api/users/users/'):
+            for method in ('post', 'put', 'patch', 'delete'):
+                with self.subTest(path=path, method=method):
+                    self.assertEqual(self._get_response_schema(path, method, '403'), ref)
+
+    def test_create_documents_error_response_for_either_shape(self):
+        """
+        A POST to a list endpoint accepts either a single object or a list, so its 400 response
+        should document both the field-keyed and the bulk error shapes.
+
+        Refs: #20054
+        """
+        for path in ('/api/dcim/sites/', '/api/ipam/prefixes/', '/api/users/users/'):
+            with self.subTest(path=path):
+                schema = self._get_response_schema(path, 'post', '400')
+                self.assertEqual(
+                    schema['oneOf'],
+                    [
+                        {'type': 'object', 'additionalProperties': {}},
+                        {'$ref': '#/components/schemas/BulkOperationError'},
+                    ],
+                )
+
+    def test_detail_operations_omit_bulk_error_response(self):
+        """
+        The bulk error body applies only to list endpoints; detail endpoints must not advertise it.
+
+        Refs: #20054
+        """
+        path = '/api/dcim/sites/{id}/'
+
+        for method in ('get', 'put', 'patch', 'delete'):
+            with self.subTest(method=method):
+                responses = self.schema['paths'][path][method]['responses']
+                self.assertNotIn('409', responses)
+                self.assertNotIn('403', responses)
+                for code, response in responses.items():
+                    schema = response.get('content', {}).get('application/json', {}).get('schema', {})
+                    self.assertNotEqual(
+                        schema.get('$ref'), '#/components/schemas/BulkOperationError',
+                        f"{method.upper()} {path} ({code}) should not reference the bulk error body"
+                    )

netbox/dcim/api/views.py

@@ -17,7 +17,6 @@ from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
 from netbox.api.metadata import ContentTypeMetadata
 from netbox.api.pagination import StripCountAnnotationsPaginator
 from netbox.api.viewsets import NetBoxModelViewSet, NetBoxReadOnlyModelViewSet
-from netbox.api.viewsets.mixins import SequentialBulkCreatesMixin
 from utilities.api import get_serializer_for_model
 from utilities.query import count_related
 from utilities.query_functions import CollateAsChar
@@ -428,7 +427,7 @@ class PlatformViewSet(NetBoxModelViewSet):
 # Devices/modules
 #
 
-class DeviceViewSet(SequentialBulkCreatesMixin, ConfigContextQuerySetMixin, RenderConfigMixin, NetBoxModelViewSet):
+class DeviceViewSet(ConfigContextQuerySetMixin, RenderConfigMixin, NetBoxModelViewSet):
     queryset = Device.objects.prefetch_related(
         'device_type__manufacturer',  # Referenced by Device.__str__() for unnamed devices
         'parent_bay',  # Referenced by DeviceSerializer.get_parent_device()

netbox/dcim/tests/test_api.py

@@ -1,7 +1,7 @@
 import json
 
 from django.conf import settings
-from django.test import tag
+from django.test import override_settings, tag
 from django.urls import reverse
 from django.utils.translation import gettext as _
 from rest_framework import status
@@ -25,6 +25,7 @@ from utilities.testing import (
     create_test_device,
     create_test_nat_ip_pair,
     disable_logging,
+    disable_warnings,
 )
 from virtualization.models import Cluster, ClusterType
 from wireless.choices import WirelessChannelChoices
@@ -467,6 +468,204 @@ class SiteTestCase(APIViewTestCases.APIViewTestCase):
         response = self.client.patch(url, data, format='json', **self.header)
         self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
 
+    def test_bulk_update_objects_non_list_body(self):
+        """
+        PATCH a list endpoint with a body which is not a list. The response should identify the
+        problem with the request as a whole, as there are no entries to report against.
+        """
+        obj_perm = ObjectPermission(name='Test permission', actions=['change'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        site = Site.objects.get(slug='site-1')
+        response = self.client.patch(
+            self._get_list_url(), {'id': site.pk, 'description': 'x'}, format='json', **self.header
+        )
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('detail', response.data)
+        self.assertNotIn('errors', response.data)
+
+        site.refresh_from_db()
+        self.assertEqual(site.description, '')
+
+        # A non-list body is described by its type, so that the client can see what was sent
+        self.assertEqual(response.data['detail'], 'Expected a list of objects, but got dict.')
+
+        # A multipart body reaches the bulk action as a QueryDict, which must be reported as the
+        # dict the client submitted rather than by that internal class name
+        response = self.client.patch(
+            self._get_list_url(), {'id': site.pk, 'description': 'x'}, format='multipart', **self.header
+        )
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(response.data['detail'], 'Expected a list of objects, but got dict.')
+
+        site.refresh_from_db()
+        self.assertEqual(site.description, '')
+
+    def test_bulk_write_objects_empty_body(self):
+        """
+        Address a list endpoint with no body at all. An absent body reaches the bulk actions as an
+        empty dict, so it must not be reported as having "got dict" -- there is no object to describe.
+        """
+        obj_perm = ObjectPermission(name='Test permission', actions=['change', 'delete'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        initial_count = Site.objects.count()
+
+        for method in ('patch', 'put', 'delete'):
+            with self.subTest(method=method):
+                response = getattr(self.client, method)(self._get_list_url(), **self.header)
+
+                self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+                self.assertEqual(
+                    response.data['detail'], 'Expected a list of objects, but no data was submitted.'
+                )
+                self.assertNotIn('errors', response.data)
+
+        # An explicitly submitted empty object is indistinguishable, and reads the same way
+        response = self.client.patch(self._get_list_url(), {}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            response.data['detail'], 'Expected a list of objects, but no data was submitted.'
+        )
+
+        self.assertEqual(Site.objects.count(), initial_count, 'No objects should have been deleted')
+
+    def test_bulk_update_objects_non_numeric_id(self):
+        """
+        PATCH a set of objects where one entry carries a non-numeric ID. The failure must be
+        correlated by position, in the same structured form as every other bulk error.
+        """
+        obj_perm = ObjectPermission(name='Test permission', actions=['change'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        site = Site.objects.get(slug='site-1')
+        data = [{'id': site.pk, 'description': 'x'}, {'id': 'not-a-number', 'description': 'y'}]
+        response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(len(response.data['errors']), 1)
+        self.assertEqual(response.data['errors'][0]['index'], 1)
+        self.assertIn('id', response.data['errors'][0]['errors'])
+
+        # The valid entry must not have been applied
+        site.refresh_from_db()
+        self.assertEqual(site.description, '')
+
+    def test_bulk_write_objects_null_entry(self):
+        """
+        Address a list endpoint with a list containing a null entry. A null fails before any field
+        is considered, so its error arrives as a bare list of messages; it must still be reported
+        as a mapping keyed by field name, as the schema declares.
+        """
+        obj_perm = ObjectPermission(name='Test permission', actions=['change', 'delete'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        site = Site.objects.get(slug='site-1')
+        initial_count = Site.objects.count()
+
+        for method in ('patch', 'put', 'delete'):
+            with self.subTest(method=method):
+                data = [{'id': site.pk, 'description': 'x'}, None]
+                response = getattr(self.client, method)(
+                    self._get_list_url(), data, format='json', **self.header
+                )
+
+                self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+                self.assertEqual(len(response.data['errors']), 1)
+                self.assertEqual(response.data['errors'][0]['index'], 1)
+
+                # Reported under the key which every other non-field error uses
+                entry_errors = response.data['errors'][0]['errors']
+                self.assertIsInstance(entry_errors, dict)
+                self.assertIn('__all__', entry_errors)
+
+        # The valid entry must not have been applied
+        site.refresh_from_db()
+        self.assertEqual(site.description, '')
+        self.assertEqual(Site.objects.count(), initial_count, 'No objects should have been deleted')
+
+    def test_bulk_update_objects_duplicate_id_invalid_entry(self):
+        """
+        PATCH a set of objects in which one object is named twice, once with invalid data and once
+        with valid data. The invalid entry must not be discarded in favor of the valid one.
+        """
+        obj_perm = ObjectPermission(name='Test permission', actions=['change'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        site = Site.objects.get(slug='site-1')
+        data = [
+            {'id': site.pk, 'name': ''},  # Invalid: name is required
+            {'id': site.pk, 'name': 'Renamed Site'},
+        ]
+        response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual([e['id'] for e in response.data['errors']], [site.pk])
+
+        # The valid entry must not have been applied
+        site.refresh_from_db()
+        self.assertEqual(site.name, 'Site 1')
+
+    def test_bulk_delete_objects_duplicate_id_changelog_message(self):
+        """
+        DELETE a set of objects in which one object is named twice with differing changelog
+        messages. The request must be rejected rather than recording only one of the messages.
+        """
+        obj_perm = ObjectPermission(name='Test permission', actions=['delete'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        site = Site.objects.get(slug='site-1')
+        data = [
+            {'id': site.pk, 'changelog_message': 'First message'},
+            {'id': site.pk, 'changelog_message': 'Second message'},
+        ]
+        response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual([e['id'] for e in response.data['errors']], [site.pk])
+        self.assertTrue(Site.objects.filter(pk=site.pk).exists())
+
+    def test_bulk_create_objects_conflicting(self):
+        """
+        POST a set of objects in which two conflict with one another. Objects are created one at a
+        time, so the second must fail validation against the first rather than passing validation
+        and then raising an IntegrityError on save.
+        """
+        obj_perm = ObjectPermission(name='Test permission', actions=['add'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        initial_count = self._get_queryset().count()
+        data = [
+            {'name': 'Site 10', 'slug': 'site-10'},
+            {'name': 'Site 11', 'slug': 'site-11'},
+            {'name': 'Site 10', 'slug': 'site-10'},  # Duplicates the first item
+        ]
+        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(self._get_queryset().count(), initial_count)
+
+        # Only the third item failed; the first two were provisionally created and rolled back
+        self.assertEqual(len(response.data['errors']), 1)
+        self.assertEqual(response.data['errors'][0]['index'], 2)
+        self.assertIn('slug', response.data['errors'][0]['errors'])
+
     def test_bulk_delete_objects_protected(self):
         """
         DELETE a set of objects where one has a protected FK dependency. Verify the structured
@@ -498,6 +697,270 @@ class SiteTestCase(APIViewTestCases.APIViewTestCase):
         self.assertTrue(Site.objects.filter(pk=site1.pk).exists(), 'Site 1 should not have been deleted')
         self.assertTrue(Site.objects.filter(pk=site2.pk).exists(), 'Site 2 should not have been deleted')
 
+    def test_bulk_update_objects_unpermitted(self):
+        """
+        PATCH a set of objects where the requesting user's object-level permissions exclude one of
+        them. The excluded object must be reported rather than silently omitted from the response.
+        """
+        site1 = Site.objects.get(slug='site-1')
+        site2 = Site.objects.get(slug='site-2')
+
+        obj_perm = ObjectPermission(name='Test permission', actions=['change'], constraints={'slug': 'site-1'})
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        data = [
+            {'id': site1.pk, 'description': 'Permitted'},
+            {'id': site2.pk, 'description': 'Not permitted'},
+        ]
+        response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('detail', response.data)
+        self.assertEqual(len(response.data['errors']), 1)
+        self.assertEqual(response.data['errors'][0]['id'], site2.pk)
+
+        # Neither site may have been updated, including the one the user is permitted to change
+        site1.refresh_from_db()
+        site2.refresh_from_db()
+        self.assertEqual(site1.description, '', 'Site 1 should not have been updated')
+        self.assertEqual(site2.description, '', 'Site 2 should not have been updated')
+
+    def test_bulk_delete_objects_unpermitted(self):
+        """
+        DELETE a set of objects where the requesting user's object-level permissions exclude one of
+        them. The excluded object must be reported rather than the request reporting success.
+        """
+        site1 = Site.objects.get(slug='site-1')
+        site2 = Site.objects.get(slug='site-2')
+
+        obj_perm = ObjectPermission(name='Test permission', actions=['delete'], constraints={'slug': 'site-1'})
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        data = [{'id': site1.pk}, {'id': site2.pk}]
+        response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('detail', response.data)
+        self.assertEqual(len(response.data['errors']), 1)
+        self.assertEqual(response.data['errors'][0]['id'], site2.pk)
+
+        # Neither site may have been deleted, including the one the user is permitted to delete
+        self.assertTrue(Site.objects.filter(pk=site1.pk).exists(), 'Site 1 should not have been deleted')
+        self.assertTrue(Site.objects.filter(pk=site2.pk).exists(), 'Site 2 should not have been deleted')
+
+    def test_bulk_update_objects_abort_request(self):
+        """
+        PATCH a set of objects where a signal receiver raises AbortRequest for more than one of
+        them. Each failure must be correlated to its own object (proving the batch continues past
+        the first abort) and no object may be modified.
+        """
+        # This tag may only be assigned to Regions, so assigning it to a Site raises AbortRequest
+        # from extras.signals.validate_assigned_tags.
+        restricted_tag = Tag.objects.create(name='Regions Only', slug='regions-only')
+        restricted_tag.object_types.set([ObjectType.objects.get_for_model(Region)])
+
+        site1 = Site.objects.get(slug='site-1')
+        site2 = Site.objects.get(slug='site-2')
+
+        obj_perm = ObjectPermission(name='Test permission', actions=['change'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+        self.add_permissions('extras.view_tag')
+
+        data = [
+            {'id': site1.pk, 'tags': [{'name': 'Regions Only'}]},
+            {'id': site2.pk, 'tags': [{'name': 'Regions Only'}]},
+        ]
+        response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('detail', response.data)
+        self.assertEqual([e['id'] for e in response.data['errors']], [site1.pk, site2.pk])
+        for error in response.data['errors']:
+            # Reported as a non-field error, in the same list-of-messages form as field errors
+            self.assertIsInstance(error['errors']['__all__'], list)
+
+        # Neither site may have been tagged (whole batch rolled back)
+        self.assertFalse(site1.tags.exists(), 'Site 1 should not have been tagged')
+        self.assertFalse(site2.tags.exists(), 'Site 2 should not have been tagged')
+
+    def test_bulk_delete_objects_abort_request(self):
+        """
+        DELETE a set of objects where a protection rule blocks more than one of them. Each failure
+        must be correlated to its own object and no object may be deleted. A protection rule is a
+        rejection of the request rather than a conflict with the state of the database, so this
+        reports 400 -- as the single-object endpoint does for the same rule.
+        """
+        site1 = Site.objects.get(slug='site-1')
+        site2 = Site.objects.get(slug='site-2')
+
+        obj_perm = ObjectPermission(name='Test permission', actions=['delete'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        # Neither site has a description, so the rule blocks both deletions via AbortRequest raised
+        # from core.signals.handle_deleted_object.
+        protection_rules = {'dcim.site': [{'description': {'required': True}}]}
+        data = [{'id': site1.pk}, {'id': site2.pk}]
+        with override_settings(PROTECTION_RULES=protection_rules):
+            response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('detail', response.data)
+        self.assertEqual([e['id'] for e in response.data['errors']], [site1.pk, site2.pk])
+        for error in response.data['errors']:
+            self.assertIsInstance(error['errors']['__all__'], list)
+
+        # Neither site may have been deleted
+        self.assertTrue(Site.objects.filter(pk=site1.pk).exists(), 'Site 1 should not have been deleted')
+        self.assertTrue(Site.objects.filter(pk=site2.pk).exists(), 'Site 2 should not have been deleted')
+
+    def test_bulk_update_objects_permission_constraint(self):
+        """
+        PATCH a set of objects where the update would move one of them outside the requesting user's
+        object-level permissions. The offending object must be named, rather than the whole batch
+        failing with an opaque 403, and nothing may be modified.
+        """
+        site1 = Site.objects.get(slug='site-1')
+        site2 = Site.objects.get(slug='site-2')
+        Site.objects.filter(pk__in=(site1.pk, site2.pk)).update(status=SiteStatusChoices.STATUS_ACTIVE)
+
+        # Only active sites may be changed, so setting Site 2's status to "planned" saves the object
+        # and then fails _validate_objects(), which perform_update() reports as PermissionDenied.
+        obj_perm = ObjectPermission(
+            name='Test permission',
+            actions=['change'],
+            constraints={'status': SiteStatusChoices.STATUS_ACTIVE},
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        data = [
+            {'id': site1.pk, 'description': 'Permitted'},
+            {'id': site2.pk, 'status': SiteStatusChoices.STATUS_PLANNED},
+        ]
+        with disable_warnings('django.request'):
+            response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+        # Still a 403, as the single-object endpoint returns, but now correlated
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+        self.assertIn('detail', response.data)
+        self.assertEqual([e['id'] for e in response.data['errors']], [site2.pk])
+        self.assertIsInstance(response.data['errors'][0]['errors']['__all__'], list)
+
+        # Neither site may have been modified, including the permitted one
+        site1.refresh_from_db()
+        site2.refresh_from_db()
+        self.assertEqual(site1.description, '', 'Site 1 should not have been updated')
+        self.assertEqual(site2.status, SiteStatusChoices.STATUS_ACTIVE, 'Site 2 should not have been updated')
+
+    def test_bulk_update_objects_permission_constraint_and_validation_error(self):
+        """
+        PATCH a set of objects where one entry is invalid and another is refused by object-level
+        permissions. Both must be reported, and the authorization failure must determine the status
+        code: it is the failure which would remain were the invalid entry corrected.
+        """
+        site1 = Site.objects.get(slug='site-1')
+        site2 = Site.objects.get(slug='site-2')
+        Site.objects.filter(pk__in=(site1.pk, site2.pk)).update(status=SiteStatusChoices.STATUS_ACTIVE)
+
+        obj_perm = ObjectPermission(
+            name='Test permission',
+            actions=['change'],
+            constraints={'status': SiteStatusChoices.STATUS_ACTIVE},
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        data = [
+            {'id': site1.pk, 'status': 'not-a-valid-status'},  # Fails validation (400)
+            {'id': site2.pk, 'status': SiteStatusChoices.STATUS_PLANNED},  # Not permitted (403)
+        ]
+        with disable_warnings('django.request'):
+            response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+        self.assertEqual([e['id'] for e in response.data['errors']], [site1.pk, site2.pk])
+
+        site1.refresh_from_db()
+        site2.refresh_from_db()
+        self.assertEqual(site1.status, SiteStatusChoices.STATUS_ACTIVE)
+        self.assertEqual(site2.status, SiteStatusChoices.STATUS_ACTIVE)
+
+    def test_bulk_create_objects_permission_constraint(self):
+        """
+        POST a set of objects where one falls outside the requesting user's object-level permissions.
+        The offending object must be correlated by its position, and nothing may be created.
+        """
+        obj_perm = ObjectPermission(
+            name='Test permission',
+            actions=['add'],
+            constraints={'status': SiteStatusChoices.STATUS_ACTIVE},
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        initial_count = self._get_queryset().count()
+        data = [
+            {'name': 'Site 20', 'slug': 'site-20', 'status': SiteStatusChoices.STATUS_ACTIVE},
+            {'name': 'Site 21', 'slug': 'site-21', 'status': SiteStatusChoices.STATUS_PLANNED},
+        ]
+        with disable_warnings('django.request'):
+            response = self.client.post(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+        self.assertEqual([e['index'] for e in response.data['errors']], [1])
+        self.assertIsInstance(response.data['errors'][0]['errors']['__all__'], list)
+
+        self.assertEqual(
+            self._get_queryset().count(), initial_count,
+            'No objects should be created when any sibling is not permitted',
+        )
+
+    def test_bulk_delete_objects_conflict_and_abort_request(self):
+        """
+        DELETE a set of objects where one is blocked by a dependent object and another by a
+        protection rule. Both failures must be reported, and the dependency conflict must determine
+        the status code: it is the failure which would remain were the request itself corrected.
+        """
+        site1 = Site.objects.get(slug='site-1')
+        site2 = Site.objects.get(slug='site-2')
+
+        obj_perm = ObjectPermission(name='Test permission', actions=['delete'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        # Site 1 is blocked by a dependent Device (ProtectedError); Site 2 is blocked by the
+        # protection rule below, as it has no description (AbortRequest). Site 1 is given a
+        # description so that only one of the two failure modes applies to it.
+        create_test_device('Protected Device', site=site1)
+        site1.description = 'Has a description'
+        site1.save()
+
+        protection_rules = {'dcim.site': [{'description': {'required': True}}]}
+        data = [{'id': site1.pk}, {'id': site2.pk}]
+        with override_settings(PROTECTION_RULES=protection_rules):
+            response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
+        self.assertEqual([e['id'] for e in response.data['errors']], [site1.pk, site2.pk])
+        for error in response.data['errors']:
+            self.assertIsInstance(error['errors']['__all__'], list)
+
+        # Neither site may have been deleted
+        self.assertTrue(Site.objects.filter(pk=site1.pk).exists(), 'Site 1 should not have been deleted')
+        self.assertTrue(Site.objects.filter(pk=site2.pk).exists(), 'Site 2 should not have been deleted')
+
 
 class LocationTestCase(APIViewTestCases.APIViewTestCase):
     model = Location
@@ -2261,10 +2724,9 @@ class DeviceTestCase(APIViewTestCases.APIViewTestCase):
 
     def test_bulk_create_objects_validation_error(self):
         """
-        POST a set of Device objects where the first passes and the second fails validation.
-        DeviceViewSet uses SequentialBulkCreatesMixin, so the response should report only the
-        failed object, and no objects should be created despite the first item passing
-        (atomic rollback).
+        POST a set of Device objects where the first passes and the second fails validation. The
+        response should report only the failed object, and no objects should be created despite
+        the first item passing (atomic rollback).
         """
         obj_perm = ObjectPermission(name='Test permission', actions=['add'])
         obj_perm.save()
@@ -2292,6 +2754,41 @@ class DeviceTestCase(APIViewTestCases.APIViewTestCase):
         self.assertEqual(response.data['errors'][0]['index'], 1)
         self.assertIn('errors', response.data['errors'][0])
 
+    def test_bulk_create_objects_abort_request(self):
+        """
+        POST a set of Device objects where a signal receiver raises AbortRequest for more than one
+        of them. Each failure must be correlated to its position in the request (proving the batch
+        continues past the first abort) and no object may be created.
+        """
+        # This tag may only be assigned to Regions, so assigning it to a Device raises AbortRequest
+        # from extras.signals.validate_assigned_tags.
+        restricted_tag = Tag.objects.create(name='Regions Only', slug='regions-only')
+        restricted_tag.object_types.set([ObjectType.objects.get_for_model(Region)])
+
+        obj_perm = ObjectPermission(name='Test permission', actions=['add'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+        self.add_permissions('extras.view_tag')
+
+        initial_count = self._get_queryset().count()
+        data = [
+            {**self.create_data[0], 'tags': [{'name': 'Regions Only'}]},
+            {**self.create_data[1], 'tags': [{'name': 'Regions Only'}]},
+        ]
+        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('detail', response.data)
+        self.assertEqual([e['index'] for e in response.data['errors']], [0, 1])
+        for error in response.data['errors']:
+            self.assertIsInstance(error['errors']['__all__'], list)
+
+        self.assertEqual(
+            self._get_queryset().count(), initial_count,
+            'No objects should be created when any sibling is aborted',
+        )
+
 
 class ModuleTestCase(APIViewTestCases.APIViewTestCase):
     model = Module

netbox/extras/tests/test_event_rules.py

@@ -19,8 +19,8 @@ from rest_framework import status
 from core.choices import JobNotificationChoices, ManagedFileRootPathChoices
 from core.events import *
 from core.models import Job, ObjectType
-from dcim.choices import SiteStatusChoices
-from dcim.models import DeviceType, Interface, Manufacturer, Site
+from dcim.choices import DeviceStatusChoices, InterfaceTypeChoices, SiteStatusChoices
+from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site
 from extras.choices import EventRuleActionChoices
 from extras.events import enqueue_event, flush_events, process_event_rules, serialize_for_event
 from extras.models import EventRule, Notification, Script, ScriptModule, Tag, Webhook
@@ -36,7 +36,8 @@ from netbox.event_rules import (
 )
 from netbox.registry import registry
 from netbox.tests.dummy_plugin.event_rules import DummyRaisingAction
-from utilities.testing import APITestCase, create_test_device
+from users.models import ObjectPermission
+from utilities.testing import APITestCase, create_test_device, disable_warnings
 from utilities.testing.mixins import RQQueueTestMixin
 
 
@@ -218,6 +219,32 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         self.assertEqual(job.kwargs['snapshots']['postchange']['name'], 'Site 1')
         self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Bar', 'Foo'])
 
+    def test_single_create_rollback_discards_events(self):
+        """
+        Check that creating an object which is then rolled back by the object-level permission check
+        in perform_create() queues no background task.
+        """
+        # Permit the creation of active sites only. The new object is saved (queueing its event)
+        # before _validate_objects() rejects it and the transaction is rolled back.
+        obj_perm = ObjectPermission(
+            name='Test permission',
+            actions=['add'],
+            constraints={'status': SiteStatusChoices.STATUS_ACTIVE},
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(Site))
+
+        data = {'name': 'Site 1', 'slug': 'site-1', 'status': SiteStatusChoices.STATUS_PLANNED}
+        url = reverse('dcim-api:site-list')
+        with disable_warnings('django.request'):
+            response = self.client.post(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+        self.assertEqual(Site.objects.count(), 0)
+
+        # No task may be queued for a creation that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     def test_bulk_create_process_eventrule(self):
         """
         Check that bulk creating multiple objects with an applicable EventRule queues a background task for each
@@ -269,6 +296,40 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
             self.assertEqual(job.kwargs['snapshots']['postchange']['name'], response.data[i]['name'])
             self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Bar', 'Foo'])
 
+    def test_bulk_create_rollback_discards_events(self):
+        """
+        Check that a sequential bulk create which is rolled back queues no background tasks for the
+        objects that were provisionally created before the failure.
+        """
+        manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
+        device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1', slug='device-type-1')
+        role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1')
+        site = Site.objects.create(name='Site 1', slug='site-1')
+
+        # Bulk creates are performed one object at a time, so each valid object is provisionally
+        # created (and its event queued) before a later object fails validation.
+        event_rule = EventRule.objects.get(name='Event Rule 1')
+        event_rule.object_types.set([ObjectType.objects.get_for_model(Device)])
+
+        data = [
+            {
+                'name': 'Device 1',
+                'device_type': device_type.pk,
+                'role': role.pk,
+                'site': site.pk,
+                'status': DeviceStatusChoices.STATUS_ACTIVE,
+            },
+            {},  # Missing all required fields
+        ]
+        url = reverse('dcim-api:device-list')
+        self.add_permissions('dcim.add_device', 'dcim.view_site', 'dcim.view_devicetype', 'dcim.view_devicerole')
+        response = self.client.post(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(Device.objects.count(), 0)
+
+        # No task may be queued for a creation that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     def test_single_update_process_eventrule(self):
         """
         Check that updating an object with an applicable EventRule queues a background task for the rule's action.
@@ -303,6 +364,37 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         self.assertEqual(job.kwargs['snapshots']['postchange']['name'], 'Site X')
         self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Baz'])
 
+    def test_single_update_rollback_discards_events(self):
+        """
+        Check that updating an object which is then rolled back by the object-level permission check
+        in perform_update() queues no background task.
+        """
+        site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE)
+
+        # Permit the modification of active sites only. Setting the status to "planned" takes the
+        # object outside the permission's scope, so it is saved (queueing its event) and then
+        # rejected by _validate_objects(), rolling the transaction back.
+        obj_perm = ObjectPermission(
+            name='Test permission',
+            actions=['change'],
+            constraints={'status': SiteStatusChoices.STATUS_ACTIVE},
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(Site))
+
+        url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
+        with disable_warnings('django.request'):
+            response = self.client.patch(
+                url, {'status': SiteStatusChoices.STATUS_PLANNED}, format='json', **self.header
+            )
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+        site.refresh_from_db()
+        self.assertEqual(site.status, SiteStatusChoices.STATUS_ACTIVE)
+
+        # No task may be queued for an update that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     def test_bulk_update_process_eventrule(self):
         """
         Check that bulk updating multiple objects with an applicable EventRule queues a background task for each
@@ -360,6 +452,38 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
             self.assertEqual(job.kwargs['snapshots']['postchange']['name'], response.data[i]['name'])
             self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Baz'])
 
+    def test_bulk_update_rollback_discards_events(self):
+        """
+        Check that a bulk update which is rolled back because one object failed validation queues no
+        background tasks for the objects that were provisionally updated.
+        """
+        sites = (
+            Site(name='Site 1', slug='site-1'),
+            Site(name='Site 2', slug='site-2'),
+            Site(name='Site 3', slug='site-3'),
+        )
+        Site.objects.bulk_create(sites)
+
+        # The first two objects are valid and will be provisionally updated; the third fails
+        # validation, rolling the entire batch back.
+        data = [
+            {'id': sites[0].pk, 'name': 'Site X'},
+            {'id': sites[1].pk, 'name': 'Site Y'},
+            {'id': sites[2].pk, 'status': 'not-a-valid-status'},
+        ]
+        url = reverse('dcim-api:site-list')
+        self.add_permissions('dcim.change_site')
+        response = self.client.patch(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+
+        # No object may have been modified
+        for site in sites:
+            site.refresh_from_db()
+        self.assertListEqual([site.name for site in sites], ['Site 1', 'Site 2', 'Site 3'])
+
+        # No task may be queued for an update that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     def test_single_delete_process_eventrule(self):
         """
         Check that deleting an object with an applicable EventRule queues a background task for the rule's action.
@@ -384,6 +508,35 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         self.assertEqual(job.kwargs['snapshots']['prechange']['name'], 'Site 1')
         self.assertEqual(job.kwargs['snapshots']['prechange']['tags'], ['Bar', 'Foo'])
 
+    def test_single_delete_rollback_discards_events(self):
+        """
+        Check that deleting an object whose cascading deletion is aborted queues no background task
+        for the dependent objects that were already processed.
+        """
+        device = create_test_device('Device 1')
+        Interface.objects.create(
+            device=device, name='Interface 1', type=InterfaceTypeChoices.TYPE_1GE_FIXED, description='Has one'
+        )
+        Interface.objects.create(device=device, name='Interface 2', type=InterfaceTypeChoices.TYPE_1GE_FIXED)
+
+        event_rule = EventRule.objects.get(name='Event Rule 3')
+        event_rule.object_types.set([ObjectType.objects.get_for_model(Interface)])
+
+        url = reverse('dcim-api:device-detail', kwargs={'pk': device.pk})
+        self.add_permissions('dcim.delete_device')
+
+        # Deleting the Device cascades to both Interfaces. The first satisfies the protection rule
+        # and so is processed (queueing its event); the second does not, aborting the request.
+        protection_rules = {'dcim.interface': [{'description': {'required': True}}]}
+        with override_settings(PROTECTION_RULES=protection_rules):
+            response = self.client.delete(url, **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertTrue(Device.objects.filter(pk=device.pk).exists())
+        self.assertEqual(Interface.objects.filter(device=device).count(), 2)
+
+        # No task may be queued for a deletion that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     def test_bulk_delete_process_eventrule(self):
         """
         Check that bulk deleting multiple objects with an applicable EventRule queues a background task for each
@@ -418,6 +571,63 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
             self.assertEqual(job.kwargs['snapshots']['prechange']['name'], sites[i].name)
             self.assertEqual(job.kwargs['snapshots']['prechange']['tags'], ['Bar', 'Foo'])
 
+    def test_bulk_delete_rollback_discards_events(self):
+        """
+        Check that a bulk delete which is rolled back because one object is protected queues no
+        background tasks for the objects that were provisionally deleted.
+        """
+        sites = (
+            Site(name='Site 1', slug='site-1'),
+            Site(name='Site 2', slug='site-2'),
+            Site(name='Site 3', slug='site-3'),
+        )
+        Site.objects.bulk_create(sites)
+
+        # A Device references the third Site, whose deletion will therefore raise a ProtectedError
+        # and roll the entire batch back.
+        create_test_device('Device 1', site=sites[2])
+
+        data = [{'id': site.pk} for site in sites]
+        url = reverse('dcim-api:site-list')
+        self.add_permissions('dcim.delete_site')
+        response = self.client.delete(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
+        self.assertEqual(Site.objects.count(), 3)
+
+        # No task may be queued for a deletion that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
+    def test_bulk_delete_abort_discards_events(self):
+        """
+        Check that a bulk delete blocked by a signal receiver raising AbortRequest (rather than by a
+        database constraint) also queues no background tasks for the objects that were provisionally
+        deleted before the failure.
+        """
+        sites = (
+            Site(name='Site 1', slug='site-1', description='Has a description'),
+            Site(name='Site 2', slug='site-2'),
+        )
+        Site.objects.bulk_create(sites)
+
+        data = [{'id': site.pk} for site in sites]
+        url = reverse('dcim-api:site-list')
+        self.add_permissions('dcim.delete_site')
+
+        # Site 2 has no description, so its deletion is blocked once Site 1 has already been deleted
+        protection_rules = {'dcim.site': [{'description': {'required': True}}]}
+        with override_settings(PROTECTION_RULES=protection_rules):
+            response = self.client.delete(url, data, format='json', **self.header)
+        # 400 rather than 409: a protection rule rejects the request, it is not a conflict with a
+        # dependent object (see BulkDestroyModelMixin.bulk_destroy)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(Site.objects.count(), 2)
+
+        # The failure is correlated to the blocked object only
+        self.assertEqual([e['id'] for e in response.data['errors']], [sites[1].pk])
+
+        # No task may be queued for a deletion that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     @skipIf('netbox.tests.dummy_plugin' not in settings.PLUGINS, 'dummy_plugin not in settings.PLUGINS')
     def test_send_webhook(self):
         request_id = uuid.uuid4()

netbox/ipam/api/views.py

@@ -19,7 +19,7 @@ from ipam import filtersets
 from ipam.models import *
 from ipam.utils import get_next_available_prefix
 from netbox.api.viewsets import NetBoxModelViewSet
-from netbox.api.viewsets.mixins import ObjectValidationMixin
+from netbox.api.viewsets.mixins import ObjectValidationMixin, discard_events_on_rollback
 from netbox.config import get_config
 from netbox.constants import ADVISORY_LOCK_KEYS
 from utilities.api import get_serializer_for_model
@@ -295,8 +295,9 @@ class AvailableObjectsView(ObjectValidationMixin, APIView):
                 return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
 
             # Create the new IP address(es)
+            using = router.db_for_write(self.queryset.model)
             try:
-                with transaction.atomic(using=router.db_for_write(self.queryset.model)):
+                with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
                     created = serializer.save()
                     self._validate_objects(created)
             except ObjectDoesNotExist:

netbox/netbox/api/serializers/bulk.py

@@ -1,11 +1,14 @@
 import copy
 import functools
 
+from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 
 from .features import ChangeLogMessageSerializer
 
 __all__ = (
+    'BulkOperationEntryErrorSerializer',
+    'BulkOperationErrorSerializer',
     'BulkOperationSerializer',
     'BulkPartialUpdateSchemaMixin',
     'BulkUpdateSchemaMixin',
@@ -17,6 +20,60 @@ class BulkOperationSerializer(ChangeLogMessageSerializer):
     id = serializers.IntegerField()
 
 
+# The two serializers below are schema-only: they are never used to validate or render data. The
+# bulk actions in netbox.api.viewsets.mixins assemble these payloads directly; these exist so that
+# their error responses are a documented part of the OpenAPI schema rather than an untyped body.
+# Note that a class docstring becomes the component's description in the published schema, so keep
+# it user-facing.
+class BulkOperationEntryErrorSerializer(serializers.Serializer):
+    """
+    The failure of a single object within a bulk operation.
+    """
+    id = serializers.IntegerField(
+        required=False,
+        help_text=_(
+            "The ID of the object which failed. Present once the entry has been matched to an "
+            "object; mutually exclusive with `index`."
+        )
+    )
+    index = serializers.IntegerField(
+        required=False,
+        help_text=_(
+            "The zero-based position of the entry within the submitted list. Used where no object "
+            "has been identified for the entry: always for creations, and for updates and deletions "
+            "where the entry itself could not be interpreted (e.g. a missing or non-numeric `id`). "
+            "Mutually exclusive with `id`."
+        )
+    )
+    errors = serializers.DictField(
+        help_text=_(
+            "The errors for this entry, keyed by field name. Values are ordinarily arrays of "
+            "messages. Errors which pertain to no particular field -- model validation, protection "
+            "rules, restricted tags, object-level permissions, or the shape of the entry itself -- "
+            "all appear under the single key `__all__`."
+        )
+    )
+
+
+class BulkOperationErrorSerializer(serializers.Serializer):
+    """
+    The body returned when a bulk operation fails, correlating each failure with the object
+    responsible for it.
+    """
+    detail = serializers.CharField(
+        help_text=_('A summary of the failure, e.g. "1 of 3 objects could not be updated."')
+    )
+    errors = BulkOperationEntryErrorSerializer(
+        many=True,
+        required=False,
+        help_text=_(
+            "One entry per object which failed; objects which would have succeeded are omitted, as "
+            "a bulk operation is all-or-none. Absent where the request could not be attributed to "
+            "individual entries at all (e.g. a request body which is not a list)."
+        )
+    )
+
+
 class BulkUpdateSchemaMixin:
     def get_fields(self):
         fields = super().get_fields()

netbox/netbox/api/viewsets/__init__.py

@@ -152,6 +152,7 @@ class NetBoxReadOnlyModelViewSet(
 class NetBoxModelViewSet(
     ETagMixin,
     mixins.BackgroundOperationMixin,
+    mixins.BulkCreateModelMixin,
     mixins.BulkUpdateModelMixin,
     mixins.BulkDestroyModelMixin,
     mixins.ObjectValidationMixin,
@@ -250,31 +251,27 @@ class NetBoxModelViewSet(
         if (response := self._handle_background_request(request, 'create')) is not None:
             return response
 
+        # Creating multiple objects, which are validated and saved one at a time in order to
+        # collect per-object errors (see BulkCreateModelMixin)
+        if isinstance(request.data, list):
+            return self.bulk_create(request, *args, **kwargs)
+
         serializer = self.get_serializer(data=request.data)
         serializer.is_valid(raise_exception=True)
-        bulk_create = getattr(serializer, 'many', False)
         self.perform_create(serializer)
 
-        # After creating the instance(s), re-initialize the serializer with a queryset
+        # After creating the instance, re-initialize the serializer with a queryset
         # to ensure related objects are prefetched.
-        if bulk_create:
-            instance_pks = [obj.pk for obj in serializer.instance]
-            # Order by PK to ensure that the ordering of objects in the response
-            # matches the ordering of those in the request.
-            qs = self.get_queryset().filter(pk__in=instance_pks).order_by('pk')
-        else:
-            qs = self.get_queryset().get(pk=serializer.instance.pk)
+        qs = self.get_queryset().get(pk=serializer.instance.pk)
 
-        # Re-serialize the instance(s) with prefetched data
-        serializer = self.get_serializer(qs, many=bulk_create)
+        # Re-serialize the instance with prefetched data
+        serializer = self.get_serializer(qs)
 
         headers = self.get_success_headers(serializer.data)
         response = Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
 
-        # Add ETag for single-object creation only (bulk returns a list, no single ETag)
-        if not bulk_create:
-            if etag := self._get_etag(qs):
-                response['ETag'] = etag
+        if etag := self._get_etag(qs):
+            response['ETag'] = etag
 
         return response
 
@@ -284,8 +281,9 @@ class NetBoxModelViewSet(
         logger.info(f"Creating new {model._meta.verbose_name}")
 
         # Enforce object-level permissions on save()
+        using = router.db_for_write(model)
         try:
-            with transaction.atomic(using=router.db_for_write(model)):
+            with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using):
                 instance = serializer.save()
                 self._validate_objects(instance)
         except ObjectDoesNotExist:
@@ -323,8 +321,9 @@ class NetBoxModelViewSet(
         logger.info(f"Updating {model._meta.verbose_name} {serializer.instance} (PK: {serializer.instance.pk})")
 
         # Enforce object-level permissions on save()
+        using = router.db_for_write(model)
         try:
-            with transaction.atomic(using=router.db_for_write(model)):
+            with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using):
                 # Re-check the If-Match ETag under a row-level lock to close the TOCTOU window
                 # between the initial check in update() and the actual write.
                 if self._get_if_match(self.request):
@@ -357,8 +356,9 @@ class NetBoxModelViewSet(
         logger = logging.getLogger(f'netbox.api.views.{self.__class__.__name__}')
         logger.info(f"Deleting {model._meta.verbose_name} {instance} (PK: {instance.pk})")
 
+        using = router.db_for_write(model)
         try:
-            with transaction.atomic(using=router.db_for_write(model)):
+            with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using):
                 # Re-check the If-Match ETag under a row-level lock to close the TOCTOU window
                 # between the initial check in destroy() and the actual delete.
                 if self._get_if_match(self.request):

netbox/netbox/api/viewsets/mixins.py

@@ -1,4 +1,7 @@
-from django.core.exceptions import ObjectDoesNotExist
+from collections import Counter
+from contextlib import contextmanager
+
+from django.core.exceptions import ObjectDoesNotExist, PermissionDenied
 from django.db import router, transaction
 from django.db.models import ProtectedError, RestrictedError
 from django.http import Http404
@@ -7,26 +10,261 @@ from rest_framework import status
 from rest_framework.exceptions import ValidationError
 from rest_framework.response import Response
 from rest_framework.reverse import reverse
+from rest_framework.settings import api_settings
 
 from core.models import ObjectType
+from core.signals import clear_events
 from extras.models import ExportTemplate
 from netbox.api.serializers import BulkOperationSerializer
 from netbox.api.serializers.bulk import get_bulk_update_serializer_class
 from netbox.jobs import AsyncAPIJob
-from utilities.exceptions import RQWorkerNotRunningException
+from utilities.exceptions import AbortRequest, RQWorkerNotRunningException
 from utilities.request import copy_safe_request
 from utilities.rqworker import any_workers_for_queue
 
 __all__ = (
+    'BULK_ERROR_STATUSES',
     'BackgroundOperationMixin',
+    'BulkCreateModelMixin',
     'BulkDestroyModelMixin',
     'BulkUpdateModelMixin',
     'CustomFieldsMixin',
     'ExportTemplatesMixin',
     'ObjectValidationMixin',
-    'SequentialBulkCreatesMixin',
+    'discard_events_on_rollback',
+    'get_duplicate_objects_response',
+    'get_invalid_entries_response',
+    'get_missing_objects_response',
+    'get_non_list_response',
+    'resolve_bulk_error_status',
+)
+
+# The status codes with which a failed bulk operation may be reported, in order of precedence: where
+# the per-object failures within one batch imply more than one of these, the earliest applies, being
+# the one which would still stand were the others corrected. An authorization failure thus outranks a
+# conflict with the current state of the database, which in turn outranks a rejection of the request.
+BULK_ERROR_STATUSES = (
+    status.HTTP_403_FORBIDDEN,
+    status.HTTP_409_CONFLICT,
+    status.HTTP_400_BAD_REQUEST,
 )
 
+PERMISSION_DENIED_MESSAGE = _("You do not have permission to perform this action on this object.")
+
+
+def resolve_bulk_error_status(error_statuses):
+    """
+    Return the single status code with which to report a bulk operation whose per-object failures
+    imply the given ones, or None if there were no failures.
+
+    :param error_statuses: The set of status codes implied by the failures within one batch, each
+        drawn from BULK_ERROR_STATUSES (which documents how they are ranked).
+    """
+    if not error_statuses:
+        return None
+
+    for error_status in BULK_ERROR_STATUSES:
+        if error_status in error_statuses:
+            return error_status
+
+    # A code with no defined precedence (a subclass may report its own) is not silently ranked;
+    # fall back to the generic client error.
+    return status.HTTP_400_BAD_REQUEST
+
+
+def get_non_list_response(data):
+    """
+    Return an error Response if the given request body is not a list of objects, or None if it is.
+
+    A bulk operation always addresses a list. The body reaching one is not necessarily a list,
+    however, as the router maps every PUT, PATCH, and DELETE on a list endpoint to a bulk action
+    regardless of what was sent. Rejecting a non-list body here keeps the per-entry errors reported
+    by the bulk actions correlated by position: those come from a serializer bound to a list, so
+    they are only positional if the body was a list to begin with.
+
+    The response carries only a `detail`, with no `errors`, as there are no entries to report
+    against.
+    """
+    if isinstance(data, list):
+        return None
+
+    if data is None or data == {} or data == '':
+        detail = _('Expected a list of objects, but no data was submitted.')
+    else:
+        # A multipart body arrives as a QueryDict rather than as a plain dict, so report any mapping
+        # by the type the client submitted rather than by the class which happens to carry it.
+        datatype = 'dict' if isinstance(data, dict) else type(data).__name__
+        detail = _('Expected a list of objects, but got {datatype}.').format(datatype=datatype)
+
+    return Response({'detail': detail}, status=status.HTTP_400_BAD_REQUEST)
+
+
+def _as_field_errors(item_errors):
+    """
+    Return the errors reported for one entry of a bulk request as a mapping of field name to messages.
+    """
+    if isinstance(item_errors, dict):
+        return item_errors
+
+    return {api_settings.NON_FIELD_ERRORS_KEY: item_errors}
+
+
+def get_invalid_entries_response(entry_errors):
+    """
+    Return a structured error Response for the entries of a bulk request which could not be
+    interpreted, or None if every entry was interpretable.
+
+    The bulk update and delete actions first check that each entry identifies an object, before any
+    entry has been matched to one. A failure at that stage -- a missing or non-numeric `id`, or an
+    entry which is not an object at all -- is reported against the entry's position in the request
+    rather than against an object ID, since no object has been identified yet. This is the same
+    correlation bulk create uses throughout, for the same reason.
+
+    Passing this stage is what allows every later error to be correlated by `id` instead.
+
+    :param entry_errors: The `errors` of a BulkOperationSerializer bound to a list, which holds one
+        entry per object in the request (an empty dict where that object was interpretable).
+    """
+    errors = [
+        {'index': i, 'errors': _as_field_errors(item_errors)}
+        for i, item_errors in enumerate(entry_errors)
+        if item_errors
+    ]
+    if not errors:
+        return None
+
+    return Response(
+        {
+            'detail': _('{failed_count} of {total} objects failed validation.').format(
+                failed_count=len(errors),
+                total=len(entry_errors),
+            ),
+            'errors': errors,
+        },
+        status=status.HTTP_400_BAD_REQUEST,
+    )
+
+
+def get_duplicate_objects_response(object_ids):
+    """
+    Return a structured error Response naming each of the given object IDs which appears more than
+    once, or None if they are all distinct.
+
+    A bulk operation identifies its objects by ID, so listing one twice is ambiguous. For an update,
+    only one of the two sets of attributes can be applied, and the discarded entry is never even
+    validated: a request pairing an invalid entry with a valid one for the same object would
+    otherwise report success while silently ignoring the invalid data. For a delete, the repetition
+    is meaningless, but it likewise causes the response to report on fewer objects than were named.
+    Rather than guess at the intent, such a request is rejected.
+    """
+    errors = [
+        {
+            'id': object_id,
+            'errors': {
+                'id': [
+                    _("Each object may be specified only once; ID {id} is listed {count} times").format(
+                        id=object_id, count=count
+                    ),
+                ],
+            },
+        }
+        # Counter preserves the order in which each ID was first seen
+        for object_id, count in Counter(object_ids).items()
+        if count > 1
+    ]
+    if not errors:
+        return None
+
+    return Response(
+        {
+            'detail': _('{failed_count} of {total} objects are listed more than once.').format(
+                failed_count=len(errors),
+                total=len(object_ids),
+            ),
+            'errors': errors,
+        },
+        status=status.HTTP_400_BAD_REQUEST,
+    )
+
+
+def get_missing_objects_response(object_ids, queryset):
+    """
+    Return a structured error Response naming each of the given object IDs which the queryset does
+    not match, or None if it matches them all.
+
+    An ID goes unmatched either because no such object exists or because the requesting user's
+    object-level permissions exclude it. The two are deliberately not distinguished, consistent with
+    the single-object endpoints, which return a 404 in both cases.
+
+    Bulk operations call this before performing any work: an unresolvable ID means the request names
+    an object the client cannot act on, so there is nothing to be gained by attempting the batch (it
+    would only be rolled back). Note that the status is 400 rather than the 409 a bulk delete
+    returns for a dependency conflict, as this is a problem with the request itself rather than with
+    the current state of the database.
+    """
+    found_pks = set(queryset.values_list('pk', flat=True))
+
+    errors = [
+        {
+            'id': object_id,
+            'errors': {
+                'id': [_("Object with ID {id} does not exist").format(id=object_id)],
+            },
+        }
+        # NetBox's bulk actions reject a repeated ID before reaching this point (see
+        # get_duplicate_objects_response), but de-duplicate anyway so that any other caller reports
+        # such an ID once rather than once per occurrence. dict.fromkeys() preserves the order of
+        # first appearance.
+        for object_id in dict.fromkeys(object_ids)
+        if object_id not in found_pks
+    ]
+    if not errors:
+        return None
+
+    return Response(
+        {
+            'detail': _('{failed_count} of {total} objects could not be found.').format(
+                failed_count=len(errors),
+                total=len(object_ids),
+            ),
+            'errors': errors,
+        },
+        status=status.HTTP_400_BAD_REQUEST,
+    )
+
+
+@contextmanager
+def discard_events_on_rollback(sender, using=None):
+    """
+    Discard any queued events if the transaction wrapping this block is rolled back.
+
+    The change logging signal receivers queue events eagerly, as the payload for a deleted object
+    must be captured while that object and its related rows are still reachable. The queue is not
+    flushed to the events pipeline until after the response has been rendered, however, so events
+    queued for writes which were subsequently rolled back would otherwise still be dispatched,
+    firing webhooks and event rules for changes that were never committed.
+
+    Bulk operations need this because they provisionally write every valid object in a batch and
+    then roll the entire batch back if any one object failed. Single-object writes need it because
+    a write can be undone after it has been saved (for instance by the object-level permission
+    check in perform_create()/perform_update(), or by a signal receiver raising AbortRequest). The
+    UI's views send the same signal when they abandon a transaction.
+
+    Must be entered *inside* the transaction whose rollback it guards, so that the rollback flag is
+    still set when this block exits. Nesting is safe: the bulk actions guard the whole batch while
+    the per-object perform_*() calls they make guard each write, and clearing an already-empty
+    queue is a no-op.
+    """
+    try:
+        yield
+    except Exception:
+        # An exception escaping the block (e.g. AbortRequest raised by a signal receiver) rolls
+        # the transaction back just as an explicit set_rollback() does.
+        clear_events.send(sender=sender)
+        raise
+    if transaction.get_connection(using).needs_rollback:
+        clear_events.send(sender=sender)
+
 
 class BackgroundOperationMixin:
     """
@@ -77,9 +315,11 @@ class BackgroundOperationMixin:
             raise RQWorkerNotRunningException()
 
         model = self.queryset.model
-        verb = _("delete") if action == 'bulk_destroy' else (
-            _("create") if action == 'create' else _("update")
-        )
+        verb = {
+            'create': _("create"),
+            'bulk_create': _("create"),
+            'bulk_destroy': _("delete"),
+        }.get(action, _("update"))
         job_name = _("Bulk {verb} {object_type}").format(
             verb=verb,
             object_type=model._meta.verbose_name_plural,
@@ -150,59 +390,102 @@ class ExportTemplatesMixin:
         return super().list(request, *args, **kwargs)
 
 
-class SequentialBulkCreatesMixin:
+class BulkCreateModelMixin:
     """
-    Perform bulk creation of new objects sequentially, rather than all at once. This ensures that any validation
-    which depends on the evaluation of existing objects (such as checking for free space within a rack) functions
-    appropriately.
+    Support the creation of multiple objects using the list endpoint for a model. Accepts a POST action with a list
+    of one or more JSON objects, each specifying the attributes of an object to be created. For example:
+
+    POST /api/dcim/sites/
+    [
+        {"name": "Site 1", "slug": "site-1"},
+        {"name": "Site 2", "slug": "site-2"}
+    ]
     """
-    def create(self, request, *args, **kwargs):
-        # If background processing was requested for a bulk (list) create, enqueue a job and
-        # return immediately. _handle_background_request() comes from BackgroundOperationMixin;
-        # fall back to "no background" so this mixin remains usable on its own (e.g. in custom
-        # viewsets).
+    def bulk_create(self, request, *args, **kwargs):
+        # If background processing was requested, enqueue a job and return immediately (before
+        # any validation, which is deferred to the worker).
         handle_background = getattr(self, '_handle_background_request', lambda *a, **kw: None)
-        if (response := handle_background(request, 'create')) is not None:
+        if (response := handle_background(request, 'bulk_create')) is not None:
             return response
 
-        # Create objects sequentially so each validation sees the state left by prior creates
-        # (e.g. rack space checks). Collect per-object errors instead of failing on the first.
-        errors = []
-        return_data = []
-        with transaction.atomic(using=router.db_for_write(self.queryset.model)):
-            if not isinstance(request.data, list):
-                # Creating a single object
-                return super().create(request, *args, **kwargs)
-
-            total = len(request.data)
-            for i, data in enumerate(request.data):
-                serializer = self.get_serializer(data=data)
-                if serializer.is_valid():
-                    # Provisionally create even when a prior item failed, so subsequent
-                    # cross-object validators (e.g. rack space checks) see a realistic state.
-                    # All creates are rolled back together if any item in the batch fails.
-                    self.perform_create(serializer)
-                    return_data.append(serializer.data)
-                else:
-                    errors.append({'index': i, 'errors': serializer.errors})
-
-            if errors:
-                transaction.set_rollback(True)
+        created_pks, errors, error_status = self.perform_bulk_create(request.data)
 
         if errors:
             return Response(
                 {
-                    'detail': _('{failed_count} of {total} objects failed validation.').format(
+                    'detail': _('{failed_count} of {total} objects could not be created.').format(
                         failed_count=len(errors),
-                        total=total,
+                        total=len(request.data),
                     ),
                     'errors': errors,
                 },
-                status=status.HTTP_400_BAD_REQUEST,
+                status=error_status,
             )
 
-        headers = self.get_success_headers(return_data[-1]) if return_data else {}
-        return Response(return_data, status=status.HTTP_201_CREATED, headers=headers)
+        # Re-fetch the new objects to serialize them with their related objects prefetched. Order by PK
+        # to ensure that the ordering of objects in the response matches the ordering of those in the
+        # request (the objects were created in the order given, so PK order is request order).
+        qs = self.get_queryset().filter(pk__in=created_pks).order_by('pk')
+        serializer = self.get_serializer(qs, many=True)
+
+        return Response(serializer.data, status=status.HTTP_201_CREATED)
+
+    def perform_bulk_create(self, data):
+        """
+        Validate and create each of the given objects, rolling the entire batch back if any one of
+        them could not be created.
+
+        Returns the PKs of the objects created, the per-object errors (if any), and the status code
+        with which to report them (None if there were none).
+        """
+        created_pks = []
+        errors = []
+        error_statuses = set()
+        using = router.db_for_write(self.queryset.model)
+        with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
+            # Validate and save each object in turn, rather than validating the entire batch up front, so that
+            # validation which depends on the state left by prior saves is evaluated correctly.
+            for i, item in enumerate(data):
+                if not isinstance(item, dict):
+                    # Checked explicitly because get_serializer() infers many=True from a list, so a nested list would
+                    # otherwise be validated as a batch of its own.
+                    errors.append({
+                        'index': i,
+                        'errors': {
+                            api_settings.NON_FIELD_ERRORS_KEY: [
+                                _('Invalid data. Expected a dictionary, but got {datatype}.').format(
+                                    datatype=type(item).__name__
+                                ),
+                            ],
+                        },
+                    })
+                    error_statuses.add(status.HTTP_400_BAD_REQUEST)
+                    continue
+                serializer = self.get_serializer(data=item)
+                if not serializer.is_valid():
+                    errors.append({'index': i, 'errors': serializer.errors})
+                    error_statuses.add(status.HTTP_400_BAD_REQUEST)
+                    continue
+                try:
+                    # Provisionally create even when a prior item failed, so subsequent cross-object validators see a
+                    # realistic state. All creates are rolled back together if any item in the batch fails.
+                    self.perform_create(serializer)
+                except AbortRequest as e:
+                    # Raised by a signal receiver rather than by validation (e.g. assigning a tag which is restricted
+                    # to other object types).
+                    errors.append({'index': i, 'errors': {'__all__': [str(e.message)]}})
+                    error_statuses.add(status.HTTP_400_BAD_REQUEST)
+                except PermissionDenied:
+                    # Raised by perform_create() when the object it saved falls outside the queryset permitted to the
+                    # requesting user. Reported per object so that the offending entry is named, but still as a 403,
+                    # which is what the single-object endpoint returns for the same rejection.
+                    errors.append({'index': i, 'errors': {'__all__': [PERMISSION_DENIED_MESSAGE]}})
+                    error_statuses.add(status.HTTP_403_FORBIDDEN)
+                else:
+                    created_pks.append(serializer.instance.pk)
+            if errors:
+                transaction.set_rollback(True)
+        return created_pks, errors, resolve_bulk_error_status(error_statuses)
 
 
 class BulkUpdateModelMixin:
@@ -238,30 +521,52 @@ class BulkUpdateModelMixin:
         if (response := handle_background(request, action)) is not None:
             return response
 
+        if (response := get_non_list_response(request.data)) is not None:
+            return response
+
+        # Check that every entry identifies an object before matching any of them to one, so that
+        # a malformed entry is reported in the same form as every other bulk error
         serializer = BulkOperationSerializer(data=request.data, many=True)
-        serializer.is_valid(raise_exception=True)
+        if not serializer.is_valid():
+            return get_invalid_entries_response(serializer.errors)
 
-        qs = self.get_bulk_update_queryset().filter(
-            pk__in=[o['id'] for o in serializer.data]
-        )
+        object_ids = [o['id'] for o in serializer.validated_data]
+
+        # Reject the batch if any object is named more than once, rather than applying only one of
+        # the entries given for it.
+        if (response := get_duplicate_objects_response(object_ids)) is not None:
+            return response
 
-        # Map update data by object ID
+        qs = self.get_bulk_update_queryset().filter(pk__in=object_ids)
+
+        # Reject the batch if any of the objects to be updated could not be found, rather than
+        # silently omitting them from the response.
+        if (response := get_missing_objects_response(object_ids, qs)) is not None:
+            return response
+
+        # Map the attributes to be set for each object by its ID, taking the IDs from the validated
+        # data rather than from the request body: the body's values have not been coerced, so an ID
+        # submitted as a string ("123") would key this map by a value which never matches the
+        # integer PK it identifies, silently discarding that entry's attributes. Each `id` is
+        # excluded here rather than popped, leaving the request data as the client sent it. zip() is
+        # strict as the two sequences necessarily correspond, every entry having been validated.
         update_data = {
-            obj.pop('id'): obj for obj in request.data
+            object_id: {k: v for k, v in item.items() if k != 'id'}
+            for object_id, item in zip(object_ids, request.data, strict=True)
         }
 
-        object_pks, errors = self.perform_bulk_update(qs, update_data, partial=partial)
+        object_pks, errors, error_status = self.perform_bulk_update(qs, update_data, partial=partial)
 
         if errors:
             return Response(
                 {
-                    'detail': _('{failed_count} of {total} objects failed validation.').format(
+                    'detail': _('{failed_count} of {total} objects could not be updated.').format(
                         failed_count=len(errors),
                         total=len(object_pks) + len(errors),
                     ),
                     'errors': errors,
                 },
-                status=status.HTTP_400_BAD_REQUEST,
+                status=error_status,
             )
 
         # Prefetch related objects for all updated instances
@@ -271,9 +576,18 @@ class BulkUpdateModelMixin:
         return Response(serializer.data, status=status.HTTP_200_OK)
 
     def perform_bulk_update(self, objects, update_data, partial):
+        """
+        Validate and apply the given attributes to each of the given objects, rolling the entire
+        batch back if any one of them could not be updated.
+
+        Returns the PKs of the objects updated, the per-object errors, and the status code with
+        which to report them (None if there were none). See resolve_bulk_error_status().
+        """
         updated_pks = []
         errors = []
-        with transaction.atomic(using=router.db_for_write(self.queryset.model)):
+        error_statuses = set()
+        using = router.db_for_write(self.queryset.model)
+        with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
             # Validate and save each object in turn so subsequent validations see the DB
             # state left by prior saves (e.g. two items renamed to the same name: the second
             # will fail validation rather than raising an integrity error on save).
@@ -282,14 +596,33 @@ class BulkUpdateModelMixin:
                 if hasattr(obj, 'snapshot'):
                     obj.snapshot()
                 serializer = self.get_serializer(obj, data=data, partial=partial)
-                if serializer.is_valid():
+                if not serializer.is_valid():
+                    errors.append({'id': obj.pk, 'errors': serializer.errors})
+                    error_statuses.add(status.HTTP_400_BAD_REQUEST)
+                    continue
+                try:
                     self.perform_update(serializer)
-                    updated_pks.append(obj.pk)
+                except AbortRequest as e:
+                    # Raised by a signal receiver rather than by validation (e.g. assigning a tag
+                    # which is restricted to other object types). perform_update() wraps its write
+                    # in its own atomic block, so the connection is rolled back to that savepoint
+                    # and the remaining objects in the batch can still be evaluated. The message is
+                    # coerced to a string because a few receivers pass an exception rather than text.
+                    errors.append({'id': obj.pk, 'errors': {'__all__': [str(e.message)]}})
+                    error_statuses.add(status.HTTP_400_BAD_REQUEST)
+                except PermissionDenied:
+                    # Raised by perform_update() when the object, as modified, falls outside the
+                    # queryset permitted to the requesting user -- so unlike the check made before
+                    # the batch begins (see get_missing_objects_response), this depends on the
+                    # attributes submitted. Reported per object so that the offending entry is
+                    # named, but still as a 403, as the single-object endpoint returns.
+                    errors.append({'id': obj.pk, 'errors': {'__all__': [PERMISSION_DENIED_MESSAGE]}})
+                    error_statuses.add(status.HTTP_403_FORBIDDEN)
                 else:
-                    errors.append({'id': obj.pk, 'errors': serializer.errors})
+                    updated_pks.append(obj.pk)
             if errors:
                 transaction.set_rollback(True)
-        return updated_pks, errors
+        return updated_pks, errors, resolve_bulk_error_status(error_statuses)
 
     def get_bulk_update_serializer_class(self, *, partial=False):
         return get_bulk_update_serializer_class(
@@ -333,19 +666,35 @@ class BulkDestroyModelMixin:
         if (response := handle_background(request, 'bulk_destroy')) is not None:
             return response
 
+        if (response := get_non_list_response(request.data)) is not None:
+            return response
+
+        # Check that every entry identifies an object before matching any of them to one, so that
+        # a malformed entry is reported in the same form as every other bulk error
         serializer = BulkOperationSerializer(data=request.data, many=True)
-        serializer.is_valid(raise_exception=True)
+        if not serializer.is_valid():
+            return get_invalid_entries_response(serializer.errors)
 
-        qs = self.get_bulk_destroy_queryset().filter(
-            pk__in=[o['id'] for o in serializer.validated_data]
-        )
+        object_ids = [o['id'] for o in serializer.validated_data]
+
+        # Reject the batch if any object is named more than once, rather than ignoring the
+        # repetition (and any changelog message attached to it) and reporting success.
+        if (response := get_duplicate_objects_response(object_ids)) is not None:
+            return response
+
+        qs = self.get_bulk_destroy_queryset().filter(pk__in=object_ids)
+
+        # Reject the batch if any of the objects to be deleted could not be found, rather than
+        # silently omitting them and reporting success.
+        if (response := get_missing_objects_response(object_ids, qs)) is not None:
+            return response
 
         # Compile any changelog messages to be recorded on the objects being deleted
         changelog_messages = {
             o['id']: o.get('changelog_message') for o in serializer.validated_data
         }
 
-        errors, total = self.perform_bulk_destroy(qs, changelog_messages)
+        errors, total, error_status = self.perform_bulk_destroy(qs, changelog_messages)
 
         if errors:
             return Response(
@@ -356,16 +705,30 @@ class BulkDestroyModelMixin:
                     ),
                     'errors': errors,
                 },
-                status=status.HTTP_409_CONFLICT,
+                status=error_status,
             )
 
         return Response(status=status.HTTP_204_NO_CONTENT)
 
     def perform_bulk_destroy(self, objects, changelog_messages=None):
+        """
+        Attempt to delete each of the given objects, rolling the entire batch back if any one of
+        them could not be deleted.
+
+        Returns the per-object errors, the number of objects processed, and the status code with
+        which to report the errors (None if there were none). A dependency conflict yields a 409, as
+        it is a conflict with the current state of the database, whereas a protection rule (or any
+        other signal receiver raising AbortRequest) yields a 400, being a rejection of the request:
+        this matches the single-object endpoint, where dispatch() maps the same exception classes to
+        the same status codes. See resolve_bulk_error_status() for how a batch hitting more than one
+        of these is resolved.
+        """
         changelog_messages = changelog_messages or {}
         errors = []
         total = 0
-        with transaction.atomic(using=router.db_for_write(self.queryset.model)):
+        error_statuses = set()
+        using = router.db_for_write(self.queryset.model)
+        with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
             for obj in objects:
                 total += 1
                 if hasattr(obj, 'snapshot'):
@@ -375,6 +738,7 @@ class BulkDestroyModelMixin:
                 try:
                     self.perform_destroy(obj)
                 except (ProtectedError, RestrictedError) as e:
+                    error_statuses.add(status.HTTP_409_CONFLICT)
                     protected = list(
                         e.protected_objects if isinstance(e, ProtectedError) else e.restricted_objects
                     )
@@ -386,14 +750,29 @@ class BulkDestroyModelMixin:
                     errors.append({
                         'id': pk,
                         'errors': {
-                            '__all__': _(
-                                'Unable to delete: {n} dependent object(s) prevent deletion.'
-                            ).format(n=len(protected)),
+                            '__all__': [
+                                _('Unable to delete: {n} dependent object(s) prevent deletion.').format(
+                                    n=len(protected)
+                                ),
+                            ],
                         },
                     })
+                except AbortRequest as e:
+                    # Raised by a signal receiver rather than by a database constraint (e.g. a
+                    # PROTECTION_RULES violation caught in core.signals.handle_deleted_object).
+                    # perform_destroy() wraps its delete in its own atomic block, so the connection
+                    # is rolled back to that savepoint and the remaining objects in the batch can
+                    # still be evaluated.
+                    errors.append({'id': pk, 'errors': {'__all__': [str(e.message)]}})
+                    error_statuses.add(status.HTTP_400_BAD_REQUEST)
+                except PermissionDenied:
+                    # Raised by perform_destroy() when the object falls outside the queryset
+                    # permitted to the requesting user (reachable via the If-Match re-check).
+                    errors.append({'id': pk, 'errors': {'__all__': [PERMISSION_DENIED_MESSAGE]}})
+                    error_statuses.add(status.HTTP_403_FORBIDDEN)
             if errors:
                 transaction.set_rollback(True)
-        return errors, total
+        return errors, total, resolve_bulk_error_status(error_statuses)
 
 
 class ObjectValidationMixin:

netbox/netbox/settings.py

@@ -809,6 +809,12 @@ REST_FRAMEWORK = {
     'DEFAULT_SCHEMA_CLASS': 'core.api.schema.NetBoxAutoSchema',
     'DEFAULT_VERSION': REST_FRAMEWORK_VERSION,
     'DEFAULT_VERSIONING_CLASS': 'rest_framework.versioning.AcceptHeaderVersioning',
+    # Align REST framework's key for errors which pertain to no particular field with Django's
+    # (django.core.exceptions.NON_FIELD_ERRORS), so that the API reports such an error under one key
+    # rather than two. Model validation errors reach a response by way of full_clean(), and so are
+    # keyed by Django; errors raised by a serializer or field are keyed by REST framework. Without
+    # this, which of the two a client must read depends on the layer which rejected the request.
+    'NON_FIELD_ERRORS_KEY': '__all__',
     'SCHEMA_COERCE_METHOD_NAMES': {
         # Default mappings
         'retrieve': 'read',

netbox/netbox/tests/test_api.py

@@ -1,11 +1,13 @@
 import uuid
 
 from django.contrib.contenttypes.models import ContentType
+from django.core.exceptions import NON_FIELD_ERRORS
 from django.db.backends.postgresql.psycopg_any import NumericRange
 from django.test import RequestFactory, TestCase
 from django.urls import reverse
 from rest_framework.exceptions import ValidationError
 from rest_framework.request import Request
+from rest_framework.settings import api_settings
 
 from dcim.api.serializers import RackSerializer
 from dcim.models import Device, Site
@@ -51,6 +53,44 @@ class AppTestCase(APITestCase):
         self.assertEqual(response.data['id'], self.user.pk)
 
 
+class NonFieldErrorKeyTestCase(APITestCase):
+    """
+    REST framework's key for errors which pertain to no particular field is configured to match
+    Django's, so that the API reports such an error under `__all__` regardless of which layer
+    rejected the request (see REST_FRAMEWORK['NON_FIELD_ERRORS_KEY'] in settings). Model validation
+    errors are keyed by Django, having reached the response by way of full_clean(); errors raised by
+    a serializer or field are keyed by REST framework.
+    """
+    def setUp(self):
+        super().setUp()
+        self.add_permissions('dcim.add_site', 'dcim.view_site', 'dcim.change_site')
+        self.url = reverse('dcim-api:site-list')
+
+    def test_setting_matches_django(self):
+        self.assertEqual(api_settings.NON_FIELD_ERRORS_KEY, NON_FIELD_ERRORS)
+
+    def test_serializer_error_uses_all_key(self):
+        """An error from REST framework's own machinery (here, a non-dictionary item)."""
+        response = self.client.post(self.url, ['not an object'], format='json', **self.header)
+
+        self.assertEqual(response.status_code, 400)
+        self.assertIn(NON_FIELD_ERRORS, response.data['errors'][0]['errors'])
+
+    def test_model_validation_error_uses_all_key(self):
+        """An error from Django's full_clean(), which uses this key of its own accord."""
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        # A Location's name must be unique within its Site, enforced by a model constraint
+        location_url = reverse('dcim-api:location-list')
+        self.add_permissions('dcim.add_location', 'dcim.view_location')
+        data = {'name': 'Location 1', 'slug': 'location-1', 'site': site.pk}
+        self.assertEqual(self.client.post(location_url, data, format='json', **self.header).status_code, 201)
+
+        response = self.client.post(location_url, data, format='json', **self.header)
+
+        self.assertEqual(response.status_code, 400)
+        self.assertIn(NON_FIELD_ERRORS, response.data)
+
+
 class RelatedObjectCountFieldTestCase(TestCase):
     """
     RelatedObjectCountFields are populated by annotations applied to a viewset's queryset, which are only

netbox/netbox/tests/test_api_background.py

@@ -14,10 +14,12 @@ from unittest.mock import patch
 from django.contrib.contenttypes.models import ContentType
 from django.test import RequestFactory
 from rest_framework import status
+from rest_framework.test import APIRequestFactory
 
 from core.choices import JobStatusChoices
 from core.exceptions import JobFailed
 from core.models import Job, ObjectChange
+from dcim.api.views import RegionViewSet
 from dcim.models import DeviceType, Manufacturer, Region
 from users.models import ObjectPermission
 from utilities.request import copy_safe_request
@@ -120,6 +122,41 @@ class BackgroundBulkWriteTests(RQQueueTestMixin, APITestCase):
         self.assertTrue(job.error)
         self.assertFalse(Region.objects.filter(slug='region-a').exists())
 
+    def test_background_bulk_create_direct_invocation(self):
+        """
+        bulk_create() honors ?background=true itself, as bulk_update() and bulk_destroy() do, so a
+        caller which reaches it without passing through NetBoxModelViewSet.create() (e.g. a custom
+        viewset) still gets background processing rather than a synchronous write.
+        """
+        self.grant('add', 'view')
+        payload = [{'name': 'Region A', 'slug': 'region-a'}]
+
+        # Apply the same minimal scaffolding as AsyncAPIJob does when it invokes an action directly
+        viewset = RegionViewSet()
+        viewset.action_map = {'post': 'bulk_create'}
+        viewset.kwargs = {}
+        viewset.args = ()
+        viewset.format_kwarg = None
+        request = viewset.initialize_request(
+            APIRequestFactory().post('/api/dcim/regions/?background=true', payload, format='json')
+        )
+        request.user = self.user
+        request.id = uuid.uuid4()  # Ordinarily set by NetBox's middleware; recorded on the changelog
+        viewset.request = request
+
+        response = viewset.bulk_create(request)
+
+        self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+        job = Job.objects.get(pk=response.data['job']['id'])
+        self.assertEqual(job.name, 'Bulk create regions')
+
+        # The worker re-invokes this same action against a request carrying no query string, so the
+        # work is performed there rather than being enqueued a second time
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+        self.assertEqual(job.data['status_code'], status.HTTP_201_CREATED)
+        self.assertTrue(Region.objects.filter(slug='region-a').exists())
+        self.assertEqual(Job.objects.count(), 1)
+
     # ------------------------------------------------------------------ update
 
     def test_background_bulk_update_patch(self):
@@ -154,8 +191,9 @@ class BackgroundBulkWriteTests(RQQueueTestMixin, APITestCase):
             self.assertEqual(r.description, 'put-bg')
 
     def test_background_bulk_update_object_permission_subset(self):
-        # Constrained to Region 1 only; bulk update of all three should update only the
-        # permitted subset and SUCCEED (matching synchronous behavior; no rollback).
+        # Constrained to Region 1 only; the other two regions cannot be resolved, so the whole
+        # batch is rejected rather than the permitted subset being updated silently (matching
+        # synchronous behavior).
         self.grant('change', 'view', constraints={'name': 'Region 1'})
         payload = [{'id': r.pk, 'description': 'subset'} for r in self.regions]
         response = self.client.patch(
@@ -163,11 +201,14 @@ class BackgroundBulkWriteTests(RQQueueTestMixin, APITestCase):
         )
         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
         job = Job.objects.get(pk=response.data['job']['id'])
-        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
-        # DB side effect AND captured result reflect the permitted subset only.
-        self.assertEqual(Region.objects.filter(description='subset').count(), 1)
-        self.assertEqual(job.data['status_code'], status.HTTP_200_OK)
-        self.assertEqual(len(job.data['data']), 1)
+        self.assertEqual(job.status, JobStatusChoices.STATUS_FAILED)
+        # Nothing was updated, and the unresolvable IDs are named in the captured result.
+        self.assertEqual(Region.objects.filter(description='subset').count(), 0)
+        self.assertEqual(job.data['status_code'], status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            [e['id'] for e in job.data['data']['errors']],
+            [self.regions[1].pk, self.regions[2].pk],
+        )
 
     def test_background_bulk_update_all_or_nothing(self):
         self.grant('change', 'view')

netbox/utilities/testing/api.py

@@ -388,6 +388,40 @@ class APIViewTestCases:
                     self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_CREATE,
                         message=changelog_message)
 
+        def test_bulk_create_objects_invalid_item(self):
+            """
+            POST a set of objects in which one item is invalid. The failure must be correlated to
+            that item's position in the request, and the entire batch must be rolled back.
+            """
+            obj_perm = ObjectPermission(
+                name='Test permission',
+                actions=['add']
+            )
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+            initial_count = self._get_queryset().count()
+
+            # A non-dictionary is used as the invalid item because it is guaranteed to fail for every
+            # model, whereas which *fields* are required varies from one model to the next.
+            response = self.client.post(
+                self._get_list_url(),
+                [self.create_data[0], 'this is not an object'],
+                format='json',
+                **self.header,
+            )
+
+            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+            self.assertEqual(
+                self._get_queryset().count(), initial_count,
+                'No objects should be created when any sibling fails validation'
+            )
+            self.assertIn('detail', response.data)
+            self.assertEqual(len(response.data['errors']), 1)
+            self.assertEqual(response.data['errors'][0]['index'], 1)
+            self.assertIn('errors', response.data['errors'][0])
+
     class UpdateObjectViewTestCase(APITestCase):
         update_data = {}
         bulk_update_data = None
@@ -538,6 +572,39 @@ class APIViewTestCases:
                     self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_UPDATE,
                         message=changelog_message)
 
+        def test_bulk_update_objects_string_id(self):
+            """
+            PATCH a set of objects whose IDs are given as strings rather than as numbers. The ID
+            field coerces such a value, so the object is identified and its attributes must be
+            applied -- rather than the entry being treated as though it carried no data.
+            """
+            if self.bulk_update_data is None:
+                self.skipTest("Bulk update data not set")
+
+            obj_perm = ObjectPermission(name='Test permission', actions=['change'])
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+            id_list = list(self._get_queryset().values_list('id', flat=True)[:2])
+            self.assertEqual(len(id_list), 2, "Insufficient number of objects to test bulk update")
+
+            # Quote only the second ID, so that a batch mixing the two forms is covered as well
+            data = [
+                {'id': id_list[0], **self.bulk_update_data},
+                {'id': str(id_list[1]), **self.bulk_update_data},
+            ]
+
+            response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+            # The attributes must have been applied to both objects. Note that the response body is
+            # deliberately not inspected: for a model whose viewset narrows its own queryset (e.g.
+            # SavedFilter, which is restricted to shared or owned objects), an update which moves an
+            # object outside that queryset succeeds but is not echoed back.
+            self.assertHttpStatus(response, status.HTTP_200_OK)
+            for instance in self._get_queryset().filter(pk__in=id_list):
+                self.assertInstanceEqual(instance, self.bulk_update_data, api=True)
+
         def test_bulk_update_objects_validation_error(self):
             """
             PATCH a set of objects where one fails validation. Verify the structured per-object error
@@ -583,6 +650,135 @@ class APIViewTestCases:
                     f'Field {field!r} of object {id_list[0]} was modified — atomic rollback may be broken',
                 )
 
+        def test_bulk_update_objects_nonexistent_id(self):
+            """
+            PATCH a set of objects where one of the IDs does not identify an existing object. Verify
+            the structured per-object error response and that no objects are modified.
+            """
+            if self.bulk_update_data is None:
+                self.skipTest('Bulk update data not set')
+
+            obj_perm = ObjectPermission(name='Test permission', actions=['change'])
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+            id_list = list(self._get_queryset().values_list('id', flat=True)[:2])
+            self.assertEqual(len(id_list), 2, 'Insufficient number of objects to test bulk update')
+            missing_id = self._get_queryset().order_by('-id').first().id + 1
+
+            data = [{'id': id, **self.bulk_update_data} for id in (*id_list, missing_id)]
+
+            # Snapshot the objects which would otherwise have been updated
+            instances_before = list(self._get_queryset().filter(pk__in=id_list))
+
+            response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+            self.assertIn('detail', response.data)
+            self.assertIn('errors', response.data)
+            self.assertEqual(len(response.data['errors']), 1)
+            self.assertEqual(response.data['errors'][0]['id'], missing_id)
+            self.assertIn('id', response.data['errors'][0]['errors'])
+
+            # The objects named alongside the missing one must not have been updated
+            for instance_before in instances_before:
+                instance_after = self._get_queryset().get(pk=instance_before.pk)
+                for field in self.bulk_update_data:
+                    if field in ('changelog_message', 'add_tags', 'remove_tags'):
+                        continue
+                    self.assertEqual(
+                        getattr(instance_after, field, None),
+                        getattr(instance_before, field, None),
+                        f'Field {field!r} of object {instance_before.pk} was modified despite an unresolvable '
+                        f'sibling ID',
+                    )
+
+        def test_bulk_update_objects_malformed_entry(self):
+            """
+            PATCH a set of objects in which one entry does not identify an object. The failure must be
+            reported in the same structured form as a per-object failure, correlated by position.
+            """
+            if self.bulk_update_data is None:
+                self.skipTest('Bulk update data not set')
+
+            obj_perm = ObjectPermission(name='Test permission', actions=['change'])
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+            instance = self._get_queryset().first()
+
+            # The second entry omits the object ID, so it cannot be matched to an object
+            data = [{'id': instance.pk, **self.bulk_update_data}, {}]
+
+            response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+            self.assertIn('detail', response.data)
+            self.assertEqual(len(response.data['errors']), 1)
+
+            # Correlated by position, as no object was identified for this entry
+            self.assertEqual(response.data['errors'][0]['index'], 1)
+            self.assertIn('id', response.data['errors'][0]['errors'])
+
+            # The valid entry must not have been applied
+            instance_after = self._get_queryset().get(pk=instance.pk)
+            for field in self.bulk_update_data:
+                if field in ('changelog_message', 'add_tags', 'remove_tags'):
+                    continue
+                self.assertEqual(
+                    getattr(instance_after, field, None),
+                    getattr(instance, field, None),
+                    f'Field {field!r} of object {instance.pk} was modified despite a malformed sibling entry',
+                )
+
+        def test_bulk_update_objects_duplicate_id(self):
+            """
+            PATCH a set of objects in which the same object is named twice. The request must be
+            rejected rather than applying only one of the entries given for that object.
+            """
+            if self.bulk_update_data is None:
+                self.skipTest('Bulk update data not set')
+
+            obj_perm = ObjectPermission(name='Test permission', actions=['change'])
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+            id_list = list(self._get_queryset().values_list('id', flat=True)[:2])
+            self.assertEqual(len(id_list), 2, 'Insufficient number of objects to test bulk update')
+
+            # Repeat the first ID at the end of the request
+            data = [{'id': id, **self.bulk_update_data} for id in (*id_list, id_list[0])]
+
+            # Snapshot the objects which would otherwise have been updated
+            instances_before = list(self._get_queryset().filter(pk__in=id_list))
+
+            response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+            self.assertIn('detail', response.data)
+            self.assertIn('errors', response.data)
+
+            # The repeated ID must be reported once, not once per occurrence
+            self.assertEqual(len(response.data['errors']), 1)
+            self.assertEqual(response.data['errors'][0]['id'], id_list[0])
+            self.assertIn('id', response.data['errors'][0]['errors'])
+
+            # No object named in the request may have been updated, including the one named only once
+            for instance_before in instances_before:
+                instance_after = self._get_queryset().get(pk=instance_before.pk)
+                for field in self.bulk_update_data:
+                    if field in ('changelog_message', 'add_tags', 'remove_tags'):
+                        continue
+                    self.assertEqual(
+                        getattr(instance_after, field, None),
+                        getattr(instance_before, field, None),
+                        f'Field {field!r} of object {instance_before.pk} was modified despite a duplicated '
+                        f'sibling ID',
+                    )
+
     class DeleteObjectViewTestCase(APITestCase):
 
         def test_delete_object_without_permission(self):
@@ -672,6 +868,133 @@ class APIViewTestCases:
                     self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_DELETE,
                         message=changelog_message)
 
+        def test_bulk_delete_objects_nonexistent_id(self):
+            """
+            DELETE a set of objects where one of the IDs does not identify an existing object. Verify
+            the structured per-object error response and that no objects are deleted.
+            """
+            obj_perm = ObjectPermission(
+                name='Test permission',
+                actions=['delete']
+            )
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+            # Target the most recently created objects to avoid triggering recursive deletions
+            id_list = list(self._get_queryset().order_by('-id').values_list('id', flat=True)[:3])
+            self.assertEqual(len(id_list), 3, 'Insufficient number of objects to test bulk deletion')
+            missing_id = max(id_list) + 1
+            data = [{'id': id} for id in (*id_list, missing_id)]
+
+            initial_count = self._get_queryset().count()
+            response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
+
+            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+            self.assertIn('detail', response.data)
+            self.assertIn('errors', response.data)
+            self.assertEqual(len(response.data['errors']), 1)
+            self.assertEqual(response.data['errors'][0]['id'], missing_id)
+            self.assertIn('id', response.data['errors'][0]['errors'])
+
+            # The objects named alongside the missing one must not have been deleted
+            self.assertEqual(self._get_queryset().count(), initial_count)
+
+        def test_bulk_delete_objects_malformed_entry(self):
+            """
+            DELETE a set of objects in which one entry does not identify an object. The failure must be
+            reported in the same structured form as a per-object failure, correlated by position.
+            """
+            obj_perm = ObjectPermission(
+                name='Test permission',
+                actions=['delete']
+            )
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+            # Target the most recently created object to avoid triggering recursive deletions
+            instance = self._get_queryset().order_by('-id').first()
+
+            # The second entry omits the object ID, so it cannot be matched to an object
+            data = [{'id': instance.pk}, {}]
+
+            initial_count = self._get_queryset().count()
+            response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
+
+            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+            self.assertIn('detail', response.data)
+            self.assertEqual(len(response.data['errors']), 1)
+
+            # Correlated by position, as no object was identified for this entry
+            self.assertEqual(response.data['errors'][0]['index'], 1)
+            self.assertIn('id', response.data['errors'][0]['errors'])
+
+            # Nothing may have been deleted
+            self.assertEqual(self._get_queryset().count(), initial_count)
+
+        def test_bulk_delete_objects_duplicate_id(self):
+            """
+            DELETE a set of objects in which the same object is named twice. The request must be
+            rejected rather than reporting success for a batch it only partly acted on.
+            """
+            obj_perm = ObjectPermission(
+                name='Test permission',
+                actions=['delete']
+            )
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+            # Target the most recently created objects to avoid triggering recursive deletions
+            id_list = list(self._get_queryset().order_by('-id').values_list('id', flat=True)[:2])
+            self.assertEqual(len(id_list), 2, 'Insufficient number of objects to test bulk deletion')
+
+            # Repeat the first ID at the end of the request
+            data = [{'id': id} for id in (*id_list, id_list[0])]
+
+            initial_count = self._get_queryset().count()
+            response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
+
+            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+            self.assertIn('detail', response.data)
+            self.assertIn('errors', response.data)
+
+            # The repeated ID must be reported once, not once per occurrence
+            self.assertEqual(len(response.data['errors']), 1)
+            self.assertEqual(response.data['errors'][0]['id'], id_list[0])
+            self.assertIn('id', response.data['errors'][0]['errors'])
+
+            # No object named in the request may have been deleted
+            self.assertEqual(self._get_queryset().count(), initial_count)
+
+        def test_bulk_delete_objects_no_body(self):
+            """
+            DELETE a list endpoint with no body at all. Nothing may be deleted -- the request names no
+            objects, so it cannot mean "all of them" -- and the response must say so intelligibly.
+            """
+            obj_perm = ObjectPermission(
+                name='Test permission',
+                actions=['delete']
+            )
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+            initial_count = self._get_queryset().count()
+            self.assertNotEqual(initial_count, 0, 'No objects exist against which to test bulk deletion')
+
+            response = self.client.delete(self._get_list_url(), **self.header)
+
+            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+            self.assertIn('detail', response.data)
+            # There are no entries to report against, so no per-object errors are returned
+            self.assertNotIn('errors', response.data)
+            self.assertEqual(
+                self._get_queryset().count(), initial_count,
+                'A bulk delete naming no objects must not delete anything'
+            )
+
     class GraphQLTestCase(APITestCase):
         graphql_auto_filter_tests = True
         graphql_auto_filter_exclude = ()