import csv from django.conf import settings from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ObjectDoesNotExist from django.test import override_settings from django.urls import reverse from django.utils.translation import gettext as _ from core.choices import ObjectChangeActionChoices from core.models import ObjectChange, ObjectType from netbox.choices import CSVDelimiterChoices, ImportFormatChoices from netbox.models.features import ChangeLoggingMixin, CustomFieldsMixin from users.models import ObjectPermission from .base import ModelTestCase from .query_counts import assert_expected_query_count from .utils import add_custom_field_data, disable_warnings, get_random_string, post_data __all__ = ( 'ModelViewTestCase', 'ViewTestCases', ) # # UI Tests # class ModelViewTestCase(ModelTestCase): """ Base TestCase for model views. Subclass to test individual views. """ def _get_base_url(self): """ Return the base format for a URL for the test's model. Override this to test for a model which belongs to a different app (e.g. testing Interfaces within the virtualization app). """ return '{}:{}_{{}}'.format( self.model._meta.app_label, self.model._meta.model_name ) def _get_url(self, action, instance=None): """ Return the URL name for a specific action and optionally a specific instance """ url_format = self._get_base_url() # If no instance was provided, assume we don't need a unique identifier if instance is None: return reverse(url_format.format(action)) return reverse(url_format.format(action), kwargs={'pk': instance.pk}) class ViewTestCases: """ We keep any TestCases with test_* methods inside a class to prevent unittest from trying to run them. """ class GetObjectViewTestCase(ModelViewTestCase): """ Retrieve a single instance. """ @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], LOGIN_REQUIRED=False) def test_get_object_anonymous(self): # Make the request as an unauthenticated user self.client.logout() ct = ContentType.objects.get_for_model(self.model) if (ct.app_label, ct.model) in settings.EXEMPT_EXCLUDE_MODELS: # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users with disable_warnings('django.request'): response = self.client.get(self._get_queryset().first().get_absolute_url()) self.assertHttpStatus(response, 302) else: response = self.client.get(self._get_queryset().first().get_absolute_url()) self.assertHttpStatus(response, 200) def test_get_object_without_permission(self): instance = self._get_queryset().first() # Try GET without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.get(instance.get_absolute_url()), 403) def test_get_object_with_permission(self): instance = self._get_queryset().first() # Add model-level permission obj_perm = ObjectPermission( name='Test permission', actions=['view'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Try GET with model-level permission self.assertHttpStatus(self.client.get(instance.get_absolute_url()), 200) def test_get_object_with_constrained_permission(self): instance1, instance2 = self._get_queryset().all()[:2] # Add object-level permission obj_perm = ObjectPermission( name='Test permission', constraints={'pk': instance1.pk}, actions=['view'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Try GET to permitted object self.assertHttpStatus(self.client.get(instance1.get_absolute_url()), 200) # Try GET to non-permitted object self.assertHttpStatus(self.client.get(instance2.get_absolute_url()), 404) class GetObjectChangelogViewTestCase(ModelViewTestCase): """ View the changelog for an instance. """ @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_get_object_changelog(self): url = self._get_url('changelog', self._get_queryset().first()) response = self.client.get(url) self.assertHttpStatus(response, 200) class CreateObjectViewTestCase(ModelViewTestCase): """ Create a single new instance. :form_data: Data to be used when creating a new object. """ form_data = {} validation_excluded_fields = [] def test_create_object_without_permission(self): # Try GET without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.get(self._get_url('add')), 403) # Try POST without permission request = { 'path': self._get_url('add'), 'data': post_data(self.form_data), } response = self.client.post(**request) with disable_warnings('django.request'): self.assertHttpStatus(response, 403) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[]) def test_create_object_with_permission(self): # Assign unconstrained permission 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)) # Try GET with model-level permission self.assertHttpStatus(self.client.get(self._get_url('add')), 200) # Add custom field data if the model supports it if issubclass(self.model, CustomFieldsMixin): add_custom_field_data(self.form_data, self.model) # If supported, add a changelog message if issubclass(self.model, ChangeLoggingMixin): if 'changelog_message' not in self.form_data: self.form_data['changelog_message'] = get_random_string(10) # Try POST with model-level permission initial_count = self._get_queryset().count() request = { 'path': self._get_url('add'), 'data': post_data(self.form_data), } self.assertHttpStatus(self.client.post(**request), 302) self.assertEqual(initial_count + 1, self._get_queryset().count()) instance = self._get_queryset().order_by('pk').last() self.assertInstanceEqual(instance, self.form_data, exclude=self.validation_excluded_fields) # Verify ObjectChange creation if issubclass(self.model, ChangeLoggingMixin): objectchanges = ObjectChange.objects.filter( changed_object_type=ContentType.objects.get_for_model(instance), changed_object_id=instance.pk ) self.assertEqual(len(objectchanges), 1) self.assertObjectChange(objectchanges[0], action=ObjectChangeActionChoices.ACTION_CREATE, message=self.form_data['changelog_message']) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[]) def test_create_object_with_constrained_permission(self): # Assign constrained permission obj_perm = ObjectPermission( name='Test permission', constraints={'pk': 0}, # Dummy permission to deny all actions=['add'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Try GET with object-level permission self.assertHttpStatus(self.client.get(self._get_url('add')), 200) # Try to create an object (not permitted) initial_count = self._get_queryset().count() request = { 'path': self._get_url('add'), 'data': post_data(self.form_data), } self.assertHttpStatus(self.client.post(**request), 200) self.assertEqual(initial_count, self._get_queryset().count()) # Check that no object was created # Update the ObjectPermission to allow creation obj_perm.constraints = {'pk__gt': 0} obj_perm.save() # Try to create an object (permitted) request = { 'path': self._get_url('add'), 'data': post_data(self.form_data), } self.assertHttpStatus(self.client.post(**request), 302) self.assertEqual(initial_count + 1, self._get_queryset().count()) instance = self._get_queryset().order_by('pk').last() self.assertInstanceEqual(instance, self.form_data, exclude=self.validation_excluded_fields) class EditObjectViewTestCase(ModelViewTestCase): """ Edit a single existing instance. :form_data: Data to be used when updating the first existing object. """ form_data = {} validation_excluded_fields = [] def test_edit_object_without_permission(self): instance = self._get_queryset().first() # Try GET without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.get(self._get_url('edit', instance)), 403) # Try POST without permission request = { 'path': self._get_url('edit', instance), 'data': post_data(self.form_data), } with disable_warnings('django.request'): self.assertHttpStatus(self.client.post(**request), 403) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[]) def test_edit_object_with_permission(self): instance = self._get_queryset().first() # Assign model-level permission 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)) # Try GET with model-level permission self.assertHttpStatus(self.client.get(self._get_url('edit', instance)), 200) # Add custom field data if the model supports it if issubclass(self.model, CustomFieldsMixin): add_custom_field_data(self.form_data, self.model) # If supported, add a changelog message if issubclass(self.model, ChangeLoggingMixin): if 'changelog_message' not in self.form_data: self.form_data['changelog_message'] = get_random_string(10) # Try POST with model-level permission request = { 'path': self._get_url('edit', instance), 'data': post_data(self.form_data), } self.assertHttpStatus(self.client.post(**request), 302) instance = self._get_queryset().get(pk=instance.pk) self.assertInstanceEqual(instance, self.form_data, exclude=self.validation_excluded_fields) # Verify ObjectChange creation if issubclass(self.model, ChangeLoggingMixin): objectchanges = ObjectChange.objects.filter( changed_object_type=ContentType.objects.get_for_model(instance), changed_object_id=instance.pk ) self.assertEqual(len(objectchanges), 1) self.assertObjectChange(objectchanges[0], action=ObjectChangeActionChoices.ACTION_UPDATE, message=self.form_data['changelog_message']) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[]) def test_edit_object_with_constrained_permission(self): instance1, instance2 = self._get_queryset().all()[:2] # Assign constrained permission obj_perm = ObjectPermission( name='Test permission', constraints={'pk': instance1.pk}, actions=['change'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Try GET with a permitted object self.assertHttpStatus(self.client.get(self._get_url('edit', instance1)), 200) # Try GET with a non-permitted object self.assertHttpStatus(self.client.get(self._get_url('edit', instance2)), 404) # Try to edit a permitted object request = { 'path': self._get_url('edit', instance1), 'data': post_data(self.form_data), } self.assertHttpStatus(self.client.post(**request), 302) instance = self._get_queryset().get(pk=instance1.pk) self.assertInstanceEqual(instance, self.form_data, exclude=self.validation_excluded_fields) # Try to edit a non-permitted object request = { 'path': self._get_url('edit', instance2), 'data': post_data(self.form_data), } self.assertHttpStatus(self.client.post(**request), 404) class DeleteObjectViewTestCase(ModelViewTestCase): """ Delete a single instance. """ def test_delete_object_without_permission(self): instance = self._get_queryset().first() # Try GET without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.get(self._get_url('delete', instance)), 403) # Try POST without permission request = { 'path': self._get_url('delete', instance), 'data': post_data({'confirm': True}), } with disable_warnings('django.request'): self.assertHttpStatus(self.client.post(**request), 403) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_delete_object_with_permission(self): instance = self._get_queryset().first() form_data = {'confirm': True} # Assign model-level permission 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)) # Try GET with model-level permission self.assertHttpStatus(self.client.get(self._get_url('delete', instance)), 200) # If supported, add a changelog message if issubclass(self.model, ChangeLoggingMixin): form_data['changelog_message'] = get_random_string(10) # Try POST with model-level permission request = { 'path': self._get_url('delete', instance), 'data': post_data(form_data), } self.assertHttpStatus(self.client.post(**request), 302) with self.assertRaises(ObjectDoesNotExist): self._get_queryset().get(pk=instance.pk) # Verify ObjectChange creation if issubclass(self.model, ChangeLoggingMixin): objectchanges = ObjectChange.objects.filter( changed_object_type=ContentType.objects.get_for_model(instance), changed_object_id=instance.pk ) self.assertEqual(len(objectchanges), 1) self.assertObjectChange(objectchanges[0], action=ObjectChangeActionChoices.ACTION_DELETE, message=form_data['changelog_message']) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_delete_object_with_constrained_permission(self): instance1, instance2 = self._get_queryset().all()[:2] # Assign object-level permission obj_perm = ObjectPermission( name='Test permission', constraints={'pk': instance1.pk}, actions=['delete'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Try GET with a permitted object self.assertHttpStatus(self.client.get(self._get_url('delete', instance1)), 200) # Try GET with a non-permitted object self.assertHttpStatus(self.client.get(self._get_url('delete', instance2)), 404) # Try to delete a permitted object request = { 'path': self._get_url('delete', instance1), 'data': post_data({'confirm': True}), } self.assertHttpStatus(self.client.post(**request), 302) with self.assertRaises(ObjectDoesNotExist): self._get_queryset().get(pk=instance1.pk) # Try to delete a non-permitted object request = { 'path': self._get_url('delete', instance2), 'data': post_data({'confirm': True}), } self.assertHttpStatus(self.client.post(**request), 404) self.assertTrue(self._get_queryset().filter(pk=instance2.pk).exists()) class ListObjectsViewTestCase(ModelViewTestCase): """ Retrieve multiple instances. """ @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], LOGIN_REQUIRED=False) def test_list_objects_anonymous(self): # Make the request as an unauthenticated user self.client.logout() ct = ContentType.objects.get_for_model(self.model) if (ct.app_label, ct.model) in settings.EXEMPT_EXCLUDE_MODELS: # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users with disable_warnings('django.request'): response = self.client.get(self._get_url('list')) self.assertHttpStatus(response, 302) else: response = self.client.get(self._get_url('list')) self.assertHttpStatus(response, 200) def test_list_objects_without_permission(self): # Try GET without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.get(self._get_url('list')), 403) def test_list_objects_with_permission(self): # Add model-level permission obj_perm = ObjectPermission( name='Test permission', actions=['view'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Try GET with model-level permission with assert_expected_query_count(self, 'list_objects_with_permission'): response = self.client.get(self._get_url('list')) self.assertHttpStatus(response, 200) def test_list_objects_with_constrained_permission(self): instance1, instance2 = self._get_queryset().all()[:2] # Add object-level permission obj_perm = ObjectPermission( name='Test permission', constraints={'pk': instance1.pk}, actions=['view'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Try GET with object-level permission response = self.client.get(self._get_url('list')) self.assertHttpStatus(response, 200) content = str(response.content) self.assertIn(instance1.get_absolute_url(), content) self.assertNotIn(instance2.get_absolute_url(), content) def test_export_objects(self): url = self._get_url('list') # Add model-level permission obj_perm = ObjectPermission( name='Test permission', actions=['view'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Test default CSV export response = self.client.get(f'{url}?export') self.assertHttpStatus(response, 200) self.assertEqual(response.get('Content-Type'), 'text/csv; charset=utf-8') # Test table-based export response = self.client.get(f'{url}?export=table') self.assertHttpStatus(response, 200) self.assertEqual(response.get('Content-Type'), 'text/csv; charset=utf-8') @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], LOGIN_REQUIRED=False) def test_export_objects_anonymous(self): # Ensure we are logged out. self.client.logout() # Some models (e.g. the users model) always require to be logged in, so we skip them here. ct = ContentType.objects.get_for_model(self.model) if (ct.app_label, ct.model) in settings.EXEMPT_EXCLUDE_MODELS: return url = self._get_url('list') # Test default CSV (or sometimes YAML) export response = self.client.get(f'{url}?export') self.assertHttpStatus(response, 200) if hasattr(self.model, 'to_yaml'): self.assertEqual(response.get('Content-Type'), 'text/yaml') else: self.assertEqual(response.get('Content-Type'), 'text/csv; charset=utf-8') # Test table-based export response = self.client.get(f'{url}?export=table') self.assertHttpStatus(response, 200) self.assertEqual(response.get('Content-Type'), 'text/csv; charset=utf-8') class CreateMultipleObjectsViewTestCase(ModelViewTestCase): """ Create multiple instances using a single form. Expects the creation of three new instances by default. :bulk_create_count: The number of objects expected to be created (default: 3). :bulk_create_data: A dictionary of data to be used for bulk object creation. """ bulk_create_count = 3 bulk_create_data = {} validation_excluded_fields = [] def test_create_multiple_objects_without_permission(self): request = { 'path': self._get_url('add'), 'data': post_data(self.bulk_create_data), } # Try POST without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.post(**request), 403) def test_create_multiple_objects_with_permission(self): initial_count = self._get_queryset().count() request = { 'path': self._get_url('add'), 'data': post_data(self.bulk_create_data), } # Assign non-constrained permission 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)) # Bulk create objects response = self.client.post(**request) self.assertHttpStatus(response, 302) self.assertEqual(initial_count + self.bulk_create_count, self._get_queryset().count()) for instance in self._get_queryset().order_by('-pk')[:self.bulk_create_count]: self.assertInstanceEqual(instance, self.bulk_create_data, exclude=self.validation_excluded_fields) def test_create_multiple_objects_with_constrained_permission(self): initial_count = self._get_queryset().count() request = { 'path': self._get_url('add'), 'data': post_data(self.bulk_create_data), } # Assign constrained permission obj_perm = ObjectPermission( name='Test permission', actions=['add'], constraints={'pk': 0} # Dummy constraint to deny all ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Attempt to make the request with unmet constraints self.assertHttpStatus(self.client.post(**request), 200) self.assertEqual(self._get_queryset().count(), initial_count) # Update the ObjectPermission to allow creation obj_perm.constraints = {'pk__gt': 0} # Dummy constraint to allow all obj_perm.save() response = self.client.post(**request) self.assertHttpStatus(response, 302) self.assertEqual(initial_count + self.bulk_create_count, self._get_queryset().count()) for instance in self._get_queryset().order_by('-pk')[: self.bulk_create_count]: self.assertInstanceEqual(instance, self.bulk_create_data, exclude=self.validation_excluded_fields) class BulkImportObjectsViewTestCase(ModelViewTestCase): """ Create multiple instances from imported data. :csv_data: CSV data for bulk import testing. Supports two formats: 1. Tuple/list format (backwards compatible): csv_data = ( "name,slug,description", "Object 1,object-1,First object", "Object 2,object-2,Second object", ) 2. Dictionary format for multiple scenarios: csv_data = { 'default': ( "name,slug,description", "Object 1,object-1,First object", ), 'with_optional_fields': ( "name,slug,description,comments", "Object 2,object-2,Second object,With comments", ) } When using dictionary format, test_bulk_import_objects_with_permission() runs each scenario as a separate subtest with clear output: test_bulk_import_objects_with_permission (scenario=default) ... ok test_bulk_import_objects_with_permission (scenario=with_optional_fields) ... ok """ csv_data = () def get_scenarios(self): return self.csv_data.keys() if isinstance(self.csv_data, dict) else ['default'] def _get_csv_data(self, scenario_name='default'): """ Get CSV data for testing. Supports both tuple/list and dictionary formats. """ if isinstance(self.csv_data, dict): if scenario_name not in self.csv_data: available = ', '.join(self.csv_data.keys()) raise ValueError(f"Scenario '{scenario_name}' not found in csv_data. Available: {available}") return '\n'.join(self.csv_data[scenario_name]) if isinstance(self.csv_data, (tuple, list)): return '\n'.join(self.csv_data) raise TypeError(f'csv_data must be a tuple, list, or dictionary, got {type(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 = { 'data': self._get_csv_data(), 'format': ImportFormatChoices.CSV, 'csv_delimiter': CSVDelimiterChoices.AUTO, } # Test GET without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.get(self._get_url('bulk_import')), 403) # Try POST without permission response = self.client.post(self._get_url('bulk_import'), data) with disable_warnings('django.request'): self.assertHttpStatus(response, 403) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[]) def test_bulk_import_objects_with_permission(self, post_import_callback=None): # Assign model-level permission once for all scenarios 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)) # Try GET with model-level permission (only once) self.assertHttpStatus(self.client.get(self._get_url('bulk_import')), 200) # Test each scenario for scenario_name in self.get_scenarios(): with self.cleanupSubTest(scenario=scenario_name): self._test_bulk_import_with_permission_scenario(scenario_name) if post_import_callback: post_import_callback(scenario_name) def _test_bulk_import_with_permission_scenario(self, scenario_name): """ Helper method to test a single bulk import scenario. """ initial_count = self._get_queryset().count() # Get CSV data for this scenario scenario_data = self._get_csv_data(scenario_name) expected_new_objects = len(scenario_data.splitlines()) - 1 data = { 'data': scenario_data, 'format': ImportFormatChoices.CSV, 'csv_delimiter': CSVDelimiterChoices.AUTO, } # If supported, add a changelog message if issubclass(self.model, ChangeLoggingMixin): data['changelog_message'] = get_random_string(10) # Test POST with permission response = self.client.post(self._get_url('bulk_import'), data) self.assertHttpStatus(response, 302) # Verify object count increase self.assertEqual(self._get_queryset().count(), initial_count + expected_new_objects) # Verify ObjectChange creation if issubclass(self.model, ChangeLoggingMixin): request_id = response.headers.get('X-Request-ID') self.assertIsNotNone(request_id, 'Unable to determine request ID from response') objectchanges = ObjectChange.objects.filter( changed_object_type=ContentType.objects.get_for_model(self.model), request_id=request_id, action=ObjectChangeActionChoices.ACTION_CREATE, ) self.assertEqual(len(objectchanges), expected_new_objects) for oc in objectchanges: self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_CREATE, message=data['changelog_message']) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_bulk_update_objects_without_change_permission(self): # Bulk import rows carrying an object ID update existing objects. This must require the 'change' # permission, matching the REST API; the 'add' permission alone must not permit updates. 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 = { 'format': ImportFormatChoices.CSV, 'data': csv_data, 'csv_delimiter': CSVDelimiterChoices.AUTO, } # Assign only the 'add' permission 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)) # Take a snapshot of the objects targeted for update reader = csv.DictReader(array, delimiter=',') check_data = list(reader) before = { line['id']: self.model.objects.get(id=line['id']) for line in check_data } # The import must be rejected with a permissions error (form re-rendered) and no object modified response = self.client.post(self._get_url('bulk_import'), data) self.assertHttpStatus(response, 200) self.assertContains(response, 'Remove the ID column to create new objects instead.') self.assertEqual(initial_count, self._get_queryset().count()) for line in check_data: obj = self.model.objects.get(id=line['id']) for attr in line: if attr == 'id': continue # Skip relational fields (FK/M2M), consistent with test_bulk_update_objects_with_permission if self.model._meta.get_field(attr).is_relation: continue self.assertEqual(getattr(obj, attr), getattr(before[line['id']], attr)) @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 = { 'format': ImportFormatChoices.CSV, 'data': csv_data, 'csv_delimiter': CSVDelimiterChoices.AUTO, } # Updating existing objects requires both 'add' (to reach the view) and 'change' (to update) obj_perm = ObjectPermission( name='Test permission', actions=['add', 'change'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Test POST with permission self.assertHttpStatus(self.client.post(self._get_url('bulk_import'), data), 302) self.assertEqual(initial_count, self._get_queryset().count()) # Verify that each object was actually updated to match the value specified in the CSV reader = csv.DictReader(array, delimiter=',') check_data = list(reader) for line in check_data: obj = self.model.objects.get(id=line["id"]) for attr, expected in line.items(): if attr == "id": continue field = self.model._meta.get_field(attr) # Skip relational fields (FK/M2M): the CSV value can't be mapped to a comparable attribute if field.is_relation: continue actual = getattr(obj, attr) # Only verify simple scalar values against the raw CSV string; skip lists and other complex # representations that the import form transforms (e.g. choice-set extra_choices). if not isinstance(actual, (str, int, float)): continue # Compare case-insensitively to tolerate values normalized on save (e.g. MAC addresses) self.assertEqual(str(actual).lower(), str(expected).lower()) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[]) def test_bulk_import_objects_with_constrained_permission(self, post_import_callback=None): # Assign constrained permission (deny all initially) obj_perm = ObjectPermission( name='Test permission', constraints={'pk': 0}, # Dummy permission to deny all actions=['add'], ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Test each scenario with constrained permissions for scenario_name in self.get_scenarios(): with self.cleanupSubTest(scenario=scenario_name): self._test_bulk_import_constrained_scenario(scenario_name, obj_perm) if post_import_callback: post_import_callback(scenario_name) def _test_bulk_import_constrained_scenario(self, scenario_name, obj_perm): """ Helper method to test a single bulk import scenario with constrained permissions. """ initial_count = self._get_queryset().count() # Get CSV data for this scenario scenario_data = self._get_csv_data(scenario_name) expected_new_objects = len(scenario_data.splitlines()) - 1 data = { 'data': scenario_data, 'format': ImportFormatChoices.CSV, 'csv_delimiter': CSVDelimiterChoices.AUTO, } # Attempt to import non-permitted objects (should fail) self.assertHttpStatus(self.client.post(self._get_url('bulk_import'), data), 200) self.assertEqual(self._get_queryset().count(), initial_count) # Update permission constraints to allow all obj_perm.constraints = {'pk__gt': 0} # Dummy permission to allow all obj_perm.save() # Import permitted objects (should succeed) self.assertHttpStatus(self.client.post(self._get_url('bulk_import'), data), 302) self.assertEqual(self._get_queryset().count(), initial_count + expected_new_objects) class BulkEditObjectsViewTestCase(ModelViewTestCase): """ Edit multiple instances. :bulk_edit_data: A dictionary of data to be used when bulk editing a set of objects. This data should differ from that used for initial object creation within setUpTestData(). """ bulk_edit_data = {} def test_bulk_edit_objects_without_permission(self): pk_list = self._get_queryset().values_list('pk', flat=True)[:3] data = { 'pk': pk_list, '_apply': True, # Form button } # Test GET without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.get(self._get_url('bulk_edit')), 403) # Try POST without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.post(self._get_url('bulk_edit'), data), 403) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[]) def test_bulk_edit_objects_with_permission(self): pk_list = list(self._get_queryset().values_list('pk', flat=True)[:3]) data = { 'pk': pk_list, '_apply': True, # Form button } # If supported, add a changelog message if issubclass(self.model, ChangeLoggingMixin): data['changelog_message'] = get_random_string(10) # Append the form data to the request data.update(post_data(self.bulk_edit_data)) # Assign model-level permission obj_perm = ObjectPermission( name='Test permission', actions=['view', 'change'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Try POST with model-level permission response = self.client.post(self._get_url('bulk_edit'), data) self.assertHttpStatus(response, 302) for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)): self.assertInstanceEqual(instance, self.bulk_edit_data) # Verify ObjectChange creation if issubclass(self.model, ChangeLoggingMixin): request_id = response.headers.get('X-Request-ID') self.assertIsNotNone(request_id, "Unable to determine request ID from response") objectchanges = ObjectChange.objects.filter( changed_object_type=ContentType.objects.get_for_model(self.model), changed_object_id__in=pk_list ) self.assertEqual(len(objectchanges), len(pk_list)) for oc in objectchanges: self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_UPDATE, message=data['changelog_message']) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[]) def test_bulk_edit_objects_with_constrained_permission(self): pk_list = list(self._get_queryset().values_list('pk', flat=True)[:3]) data = { 'pk': pk_list, '_apply': True, # Form button } # Append the form data to the request data.update(post_data(self.bulk_edit_data)) # Dynamically determine a constraint that will *not* be matched by the updated objects. attr_name = list(self.bulk_edit_data.keys())[0] field = self.model._meta.get_field(attr_name) value = field.value_from_object(self._get_queryset().first()) # Assign constrained permission obj_perm = ObjectPermission( name='Test permission', constraints={attr_name: value}, actions=['view', 'change'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Attempt to bulk edit permitted objects into a non-permitted state response = self.client.post(self._get_url('bulk_edit'), data) self.assertHttpStatus(response, 200) # Update permission constraints obj_perm.constraints = {'pk__gt': 0} obj_perm.save() # Bulk edit permitted objects self.assertHttpStatus(self.client.post(self._get_url('bulk_edit'), data), 302) for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)): self.assertInstanceEqual(instance, self.bulk_edit_data) class BulkDeleteObjectsViewTestCase(ModelViewTestCase): """ Delete multiple instances. """ def test_bulk_delete_objects_without_permission(self): pk_list = self._get_queryset().values_list('pk', flat=True)[:3] data = { 'pk': pk_list, 'confirm': True, '_confirm': True, # Form button } # Test GET without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.get(self._get_url('bulk_delete')), 403) # Try POST without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.post(self._get_url('bulk_delete'), data), 403) def test_bulk_delete_objects_with_permission(self): pk_list = list(self._get_queryset().values_list('pk', flat=True))[:3] data = { 'pk': pk_list, 'confirm': True, '_confirm': True, # Form button } # If supported, add a changelog message if issubclass(self.model, ChangeLoggingMixin): data['changelog_message'] = get_random_string(10) # Assign unconstrained permission 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)) # Try POST with model-level permission response = self.client.post(self._get_url('bulk_delete'), data) self.assertHttpStatus(response, 302) self.assertFalse(self._get_queryset().filter(pk__in=pk_list).exists()) # Verify ObjectChange creation if issubclass(self.model, ChangeLoggingMixin): objectchanges = ObjectChange.objects.filter( changed_object_type=ContentType.objects.get_for_model(self.model), changed_object_id__in=pk_list ) self.assertEqual(len(objectchanges), len(pk_list)) for oc in objectchanges: self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_DELETE, message=data['changelog_message']) def test_bulk_delete_objects_with_constrained_permission(self): pk_list = self._get_queryset().values_list('pk', flat=True) data = { 'pk': pk_list, 'confirm': True, '_confirm': True, # Form button } # Assign constrained permission obj_perm = ObjectPermission( name='Test permission', constraints={'pk': 0}, # Dummy permission to deny all actions=['delete'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Attempt to bulk delete non-permitted objects initial_count = self._get_queryset().count() self.assertHttpStatus(self.client.post(self._get_url('bulk_delete'), data), 302) self.assertEqual(self._get_queryset().count(), initial_count) # Update permission constraints obj_perm.constraints = {'pk__gt': 0} # Dummy permission to allow all obj_perm.save() # Bulk delete permitted objects self.assertHttpStatus(self.client.post(self._get_url('bulk_delete'), data), 302) self.assertEqual(self._get_queryset().count(), 0) class BulkRenameObjectsViewTestCase(ModelViewTestCase): """ Rename multiple instances. """ rename_data = { 'find': '^(.*)$', 'replace': '\\1X', # Append an X to the original value 'use_regex': True, } def test_bulk_rename_objects_without_permission(self): pk_list = self._get_queryset().values_list('pk', flat=True)[:3] data = { 'pk': pk_list, '_apply': True, # Form button } data.update(self.rename_data) # Test GET without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.get(self._get_url('bulk_rename')), 403) # Try POST without permission with disable_warnings('django.request'): self.assertHttpStatus(self.client.post(self._get_url('bulk_rename'), data), 403) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_bulk_rename_objects_with_permission(self): objects = self._get_queryset().all()[:3] pk_list = [obj.pk for obj in objects] data = { 'pk': pk_list, '_apply': True, # Form button 'field_names': ['name'], } data.update(self.rename_data) # Assign model-level permission 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)) # Try POST with model-level permission self.assertHttpStatus(self.client.post(self._get_url('bulk_rename'), data), 302) for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)): self.assertEqual(instance.name, f'{objects[i].name}X') @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_bulk_rename_objects_with_changelog_message(self): if not issubclass(self.model, ChangeLoggingMixin): self.skipTest("Model does not support change logging") objects = self._get_queryset().all()[:3] pk_list = [obj.pk for obj in objects] data = { 'pk': pk_list, '_apply': True, 'changelog_message': 'Bulk rename test message', 'field_names': ['name'], } data.update(self.rename_data) # Assign model-level permission 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.assertHttpStatus(self.client.post(self._get_url('bulk_rename'), data), 302) # Verify changelog message was recorded on each renamed object object_type = ObjectType.objects.get_for_model(self.model) for pk in pk_list: oc = ObjectChange.objects.filter( changed_object_type=object_type, changed_object_id=pk, action=ObjectChangeActionChoices.ACTION_UPDATE, ).order_by('-time').first() self.assertIsNotNone(oc) self.assertEqual(oc.message, 'Bulk rename test message') @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_bulk_rename_objects_with_constrained_permission(self): objects = self._get_queryset().all()[:3] pk_list = [obj.pk for obj in objects] data = { 'pk': pk_list, '_apply': True, # Form button 'field_names': ['name'], } data.update(self.rename_data) # Assign constrained permission obj_perm = ObjectPermission( name='Test permission', constraints={'name__regex': '[^X]$'}, actions=['change'] ) obj_perm.save() obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) # Attempt to bulk edit permitted objects into a non-permitted state response = self.client.post(self._get_url('bulk_rename'), data) self.assertHttpStatus(response, 200) # Update permission constraints obj_perm.constraints = {'pk__gt': 0} obj_perm.save() # Bulk rename permitted objects self.assertHttpStatus(self.client.post(self._get_url('bulk_rename'), data), 302) for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)): self.assertEqual(instance.name, f'{objects[i].name}X') @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_bulk_rename_label_field(self): """When field_names=['label'] is submitted, labels (not names) are updated.""" if 'label' not in {f.name for f in self.model._meta.fields}: self.skipTest("Model does not have a label field") objects = self._get_queryset().all()[:3] pk_list = [obj.pk for obj in objects] original_labels = [obj.label for obj in objects] data = { 'pk': pk_list, 'field_names': ['label'], '_apply': True, } data.update(self.rename_data) 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.assertHttpStatus(self.client.post(self._get_url('bulk_rename'), data), 302) for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)): self.assertEqual(instance.label, f'{original_labels[i]}X') self.assertEqual(instance.name, objects[i].name) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_bulk_rename_name_and_label_fields(self): """When field_names=['name', 'label'] is submitted, both fields are updated simultaneously.""" if 'label' not in {f.name for f in self.model._meta.fields}: self.skipTest("Model does not have a label field") objects = self._get_queryset().all()[:3] pk_list = [obj.pk for obj in objects] original_names = [obj.name for obj in objects] original_labels = [obj.label for obj in objects] data = { 'pk': pk_list, 'field_names': ['name', 'label'], '_apply': True, } data.update(self.rename_data) 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.assertHttpStatus(self.client.post(self._get_url('bulk_rename'), data), 302) for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)): self.assertEqual(instance.name, f'{original_names[i]}X') self.assertEqual(instance.label, f'{original_labels[i]}X') class PrimaryObjectViewTestCase( GetObjectViewTestCase, GetObjectChangelogViewTestCase, CreateObjectViewTestCase, EditObjectViewTestCase, DeleteObjectViewTestCase, ListObjectsViewTestCase, BulkImportObjectsViewTestCase, BulkEditObjectsViewTestCase, BulkDeleteObjectsViewTestCase, ): """ TestCase suitable for testing all standard View functions for primary objects """ maxDiff = None class OrganizationalObjectViewTestCase( GetObjectViewTestCase, GetObjectChangelogViewTestCase, CreateObjectViewTestCase, EditObjectViewTestCase, DeleteObjectViewTestCase, ListObjectsViewTestCase, BulkImportObjectsViewTestCase, BulkEditObjectsViewTestCase, BulkDeleteObjectsViewTestCase, ): """ TestCase suitable for all organizational objects """ maxDiff = None class AdminModelViewTestCase( GetObjectViewTestCase, CreateObjectViewTestCase, EditObjectViewTestCase, DeleteObjectViewTestCase, ListObjectsViewTestCase, BulkImportObjectsViewTestCase, BulkEditObjectsViewTestCase, BulkDeleteObjectsViewTestCase, ): """ TestCase suitable for testing all standard View functions for objects which inherit from AdminModel. """ maxDiff = None class DeviceComponentTemplateViewTestCase( EditObjectViewTestCase, DeleteObjectViewTestCase, CreateMultipleObjectsViewTestCase, BulkEditObjectsViewTestCase, BulkRenameObjectsViewTestCase, BulkDeleteObjectsViewTestCase, ): """ TestCase suitable for testing device component template models (ConsolePortTemplates, InterfaceTemplates, etc.) """ maxDiff = None class DeviceComponentViewTestCase( GetObjectViewTestCase, GetObjectChangelogViewTestCase, EditObjectViewTestCase, DeleteObjectViewTestCase, ListObjectsViewTestCase, CreateMultipleObjectsViewTestCase, BulkImportObjectsViewTestCase, BulkEditObjectsViewTestCase, BulkRenameObjectsViewTestCase, BulkDeleteObjectsViewTestCase, ): """ TestCase suitable for testing device component models (ConsolePorts, Interfaces, etc.) """ maxDiff = None