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
