add migration for django 4, fix serializer validation (#258)

changes/258.added

@@ -0,0 +1 @@
+Added migration to support Django 4.

changes/258.fixed

@@ -0,0 +1 @@
+Fixed IPRangeSerializer requiring vrf field.

nautobot_firewall_models/api/serializers.py

@@ -1,4 +1,5 @@
 """API serializers for firewall models."""
+
 from rest_framework import serializers
 from nautobot.apps.api import NautobotModelSerializer, ValidatedModelSerializer
 
@@ -17,6 +18,37 @@ class Meta:
         model = models.IPRange
         fields = "__all__"
 
+        # Omit the UniqueTogetherValidators that would be automatically added to validate (start_address, end_address, vrf).
+        # This prevents vrf from being interpreted as a required field.
+        validators = []
+
+    def validate(self, data):
+        """Custom validate method to enforce unique constraints on IPRange model."""
+        # Validate uniqueness of (start_address, end_address, vrf) since we omitted the automatically-created validator above.
+        start_address = data.get("start_address")
+        end_address = data.get("end_address")
+        vrf = data.get("vrf")
+        if not any([start_address is not None, end_address is not None, vrf is not None]):
+            return super().validate(data)
+
+        # Use existing object's attributes for partial updates
+        if self.instance:
+            start_address = start_address or self.instance.start_address
+            end_address = end_address or self.instance.end_address
+            vrf = vrf or self.instance.vrf
+            qs = models.IPRange.objects.exclude(pk=self.instance.pk)
+        else:
+            qs = models.IPRange.objects.all()
+
+        if vrf is not None:
+            if qs.filter(start_address=start_address, end_address=end_address, vrf=vrf).exists():
+                raise serializers.ValidationError("The fields start_address, end_address, vrf must make a unique set.")
+        else:
+            if qs.filter(start_address=start_address, end_address=end_address, vrf__isnull=True).exists():
+                raise serializers.ValidationError("The fields start_address, end_address must make a unique set.")
+
+        return super().validate(data)
+
 
 class FQDNSerializer(NautobotModelSerializer):
     """FQDN Serializer."""

nautobot_firewall_models/jobs.py

@@ -1,4 +1,5 @@
 """Jobs to run backups, intended config, and compliance."""
+
 from nautobot.extras.jobs import Job, MultiObjectVar, get_task_logger
 from nautobot.core.celery import register_jobs
 
@@ -49,7 +50,7 @@ def run(self, device):  # pylint: disable=arguments-differ
             device_obj = Device.objects.get(pk=dev)
             logger.debug("Running against Device: `%s`", str(device_obj))
             CapircaPolicy.objects.update_or_create(device=device_obj)
-            logger.info(f"{device_obj} Updated", extra={"object": device_obj})
+            logger.info("%s Updated", device_obj, extra={"object": device_obj})
 
 
 jobs = [RunCapircaJob]

nautobot_firewall_models/migrations/0021_alter_addressobject_status_and_more.py

@@ -0,0 +1,165 @@
+# Generated by Django 4.2.14 on 2024-08-06 00:36
+
+from django.db import migrations
+import django.db.models.deletion
+import nautobot.extras.models.statuses
+import nautobot_firewall_models.utils
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("nautobot_firewall_models", "0020_field_cleanups"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="addressobject",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="addressobjectgroup",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="applicationobject",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="applicationobjectgroup",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="fqdn",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="iprange",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="natpolicy",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="natpolicyrule",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="policy",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="policyrule",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="serviceobject",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="serviceobjectgroup",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="userobject",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="userobjectgroup",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="zone",
+            name="status",
+            field=nautobot.extras.models.statuses.StatusField(
+                default=nautobot_firewall_models.utils.get_default_status,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="%(app_label)s_%(class)s_related",
+                to="extras.status",
+            ),
+        ),
+    ]

nautobot_firewall_models/tests/test_api.py

@@ -1,11 +1,15 @@
 """Unit tests for API views."""
+
 # flake8: noqa: F403,405
 # pylint: disable=invalid-name
 # pylint: disable=duplicate-code
-from nautobot.apps.testing import APIViewTestCases
+from django.contrib.contenttypes.models import ContentType
+from nautobot.apps.testing import APIViewTestCases, disable_warnings
 from nautobot.dcim.models import Location, Platform, DeviceType, Device
 from nautobot.extras.models import Status, Role
-from nautobot.ipam.models import Prefix
+from nautobot.ipam.models import Prefix, VRF
+from nautobot.users.models import ObjectPermission
+from rest_framework import status as drf_status
 
 from nautobot_firewall_models import models
 from . import fixtures
@@ -27,6 +31,90 @@ def setUpTestData(cls):
         ]
         fixtures.create_ip_range()
 
+    def test_unique_validators(self):
+        """Test the unique validators for IPRange."""
+        # Add object-level permission
+        obj_perm = ObjectPermission(name="Test permission", actions=["add", "change"])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))
+
+        vrfs = (
+            VRF.objects.create(name="test vrf 1"),
+            VRF.objects.create(name="test vrf 2"),
+        )
+
+        url = self._get_list_url()
+
+        # Create an IPRange object with a vrf
+        models.IPRange.objects.create(start_address="1.0.0.1", end_address="1.0.0.8", vrf=vrfs[0])
+
+        initial_count = self._get_queryset().count()
+
+        # Create an IPRange object with the same start and end address but a different vrf
+        with disable_warnings("django.request"):
+            data = {"start_address": "1.0.0.1", "end_address": "1.0.0.8", "vrf": vrfs[1].pk}
+            response = self.client.post(url, data, format="json", **self.header)
+            self.assertHttpStatus(response, drf_status.HTTP_201_CREATED)
+            self.assertEqual(self._get_queryset().count(), initial_count + 1)
+
+        # Creating an IPRange object with the same start and end address and the same vrf fails
+        with disable_warnings("django.request"):
+            data = {"start_address": "1.0.0.1", "end_address": "1.0.0.8", "vrf": vrfs[0].pk}
+            response = self.client.post(url, data, format="json", **self.header)
+            self.assertHttpStatus(response, drf_status.HTTP_400_BAD_REQUEST)
+            self.assertIn("non_field_errors", response.data)
+            self.assertEqual(
+                "The fields start_address, end_address, vrf must make a unique set.",
+                response.data["non_field_errors"][0],
+            )
+            self.assertEqual(self._get_queryset().count(), initial_count + 1)
+
+        # Create another IPRange object with no vrf
+        models.IPRange.objects.create(start_address="2.0.0.1", end_address="2.0.0.8")
+
+        # Creating an IPRange object with the same start and end address with no vrf fails
+        with disable_warnings("django.request"):
+            data = {"start_address": "2.0.0.1", "end_address": "2.0.0.8"}
+            response = self.client.post(url, data, format="json", **self.header)
+            self.assertHttpStatus(response, drf_status.HTTP_400_BAD_REQUEST)
+            self.assertIn("non_field_errors", response.data)
+            self.assertEqual(
+                "The fields start_address, end_address must make a unique set.",
+                response.data["non_field_errors"][0],
+            )
+            self.assertEqual(self._get_queryset().count(), initial_count + 2)
+
+        # Create an IPRange object with the same start and end address with a vrf
+        with disable_warnings("django.request"):
+            data = {"start_address": "2.0.0.1", "end_address": "2.0.0.8", "vrf": vrfs[0].pk}
+            response = self.client.post(url, data, format="json", **self.header)
+            self.assertHttpStatus(response, drf_status.HTTP_201_CREATED)
+            self.assertEqual(self._get_queryset().count(), initial_count + 3)
+
+        # Patching an existing object to violate uniqueness constraints also fails validation
+        ip_ranges = (
+            models.IPRange.objects.create(start_address="123.0.0.1", end_address="123.0.0.10"),
+            models.IPRange.objects.create(start_address="123.0.0.11", end_address="123.0.0.20"),
+        )
+        with disable_warnings("django.request"):
+            data = {"start_address": ip_ranges[0].start_address, "end_address": ip_ranges[0].end_address}
+            url = self._get_detail_url(ip_ranges[1])
+
+            response = self.client.patch(url, data, format="json", **self.header)
+            self.assertHttpStatus(response, drf_status.HTTP_400_BAD_REQUEST)
+            self.assertIn("non_field_errors", response.data)
+            self.assertEqual(
+                "The fields start_address, end_address must make a unique set.",
+                response.data["non_field_errors"][0],
+            )
+
+            # Using a different vrf works
+            data["vrf"] = vrfs[0].pk
+            response = self.client.patch(url, data, format="json", **self.header)
+            self.assertHttpStatus(response, drf_status.HTTP_200_OK)
+            self.assertEqual(response.data["vrf"]["id"], vrfs[0].pk)
+
 
 class FQDNAPIViewTest(APIViewTestCases.APIViewTestCase):
     """Test the Protocol viewsets."""

nautobot_firewall_models/tests/test_ui_views.py

@@ -1,4 +1,5 @@
 """Unit tests for views."""
+
 # flake8: noqa: F403,405
 # pylint: disable=invalid-name
 # pylint: disable=duplicate-code