7961 CSV bulk update (#10715)

* 7961 add csv bulk update

* temp checkin - blocked

* 7961 bugfix and cleanup

* 7961 change to id, add docs

* 7961 add tests cases

* 7961 fix does not exist validation error

* 7961 fix does not exist validation error

* 7961 update tests

* 7961 update tests

* 7961 update tests

* 7961 update tests

* 7961 update tests

* 7961 update tests

* 7961 update tests

* 7961 update tests

* 7961 update tests

* 7961 make test cases more explicit

* 7961 make test cases more explicit

* 7961 make test cases more explicit

* 7961 make test cases more explicit

* 7961 make test cases more explicit

* 7961 make test cases more explicit

* 7961 make test cases more explicit

* 7961 optimize loading csv test data

* 7961 update tests remove redundant code

* 7961 avoid MPTT issue in test cases

netbox/utilities/forms/utils.py

@@ -220,7 +220,11 @@ def validate_csv(headers, fields, required_fields):
     if parsed csv data contains invalid headers or does not contain required headers.
     """
     # Validate provided column headers
+    is_update = False
     for field, to_field in headers.items():
+        if field == "id":
+            is_update = True
+            continue
         if field not in fields:
             raise forms.ValidationError(f'Unexpected column header "{field}" found.')
         if to_field and not hasattr(fields[field], 'to_field_name'):
@@ -228,7 +232,8 @@ def validate_csv(headers, fields, required_fields):
         if to_field and not hasattr(fields[field].queryset.model, to_field):
             raise forms.ValidationError(f'Invalid related object attribute for column "{field}": {to_field}')
 
-    # Validate required fields
-    for f in required_fields:
-        if f not in headers:
-            raise forms.ValidationError(f'Required column header "{f}" not found.')
+    # Validate required fields (if not an update)
+    if not is_update:
+        for f in required_fields:
+            if f not in headers:
+                raise forms.ValidationError(f'Required column header "{f}" not found.')

netbox/utilities/testing/views.py

@@ -1,5 +1,8 @@
+import csv
+
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ObjectDoesNotExist
+from django.db.models import ForeignKey
 from django.test import override_settings
 from django.urls import reverse
 
@@ -19,6 +22,7 @@ __all__ = (
 # UI Tests
 #
 
+
 class ModelViewTestCase(ModelTestCase):
     """
     Base TestCase for model views. Subclass to test individual views.
@@ -546,6 +550,9 @@ class ViewTestCases:
         def _get_csv_data(self):
             return '\n'.join(self.csv_data)
 
+        def _get_update_csv_data(self):
+            return self.csv_update_data, '\n'.join(self.csv_update_data)
+
         def test_bulk_import_objects_without_permission(self):
             data = {
                 'csv': self._get_csv_data(),
@@ -583,6 +590,42 @@ class ViewTestCases:
             self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200)
             self.assertEqual(self._get_queryset().count(), initial_count + len(self.csv_data) - 1)
 
+        @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
+        def test_bulk_update_objects_with_permission(self):
+            if not hasattr(self, 'csv_update_data'):
+                raise NotImplementedError("The test must define csv_update_data.")
+
+            initial_count = self._get_queryset().count()
+            array, csv_data = self._get_update_csv_data()
+            data = {
+                'csv': csv_data,
+            }
+
+            # Assign model-level permission
+            obj_perm = ObjectPermission(
+                name='Test permission',
+                actions=['add']
+            )
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))
+
+            # Test POST with permission
+            self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200)
+            self.assertEqual(initial_count, self._get_queryset().count())
+
+            reader = csv.DictReader(array, delimiter=',')
+            check_data = list(reader)
+            for line in check_data:
+                obj = self.model.objects.get(id=line["id"])
+                for attr, value in line.items():
+                    if attr != "id":
+                        field = self.model._meta.get_field(attr)
+                        value = getattr(obj, attr)
+                        # cannot verify FK fields as don't know what name the CSV maps to
+                        if value is not None and not isinstance(field, ForeignKey):
+                            self.assertEqual(value, value)
+
         @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
         def test_bulk_import_objects_with_constrained_permission(self):
             initial_count = self._get_queryset().count()