Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add migration for django 4, fix serializer validation #258

Merged
merged 6 commits into from
Aug 7, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changes/258.added
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Added migration to support Django 4.
1 change: 1 addition & 0 deletions changes/258.fixed
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fixed IPRangeSerializer requiring vrf field.
32 changes: 32 additions & 0 deletions nautobot_firewall_models/api/serializers.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""API serializers for firewall models."""

from rest_framework import serializers
from nautobot.apps.api import NautobotModelSerializer, ValidatedModelSerializer

Expand All @@ -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."""
Expand Down
3 changes: 2 additions & 1 deletion nautobot_firewall_models/jobs.py
Original file line number Diff line number Diff line change
@@ -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

Expand Down Expand Up @@ -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]
Expand Down
Original file line number Diff line number Diff line change
@@ -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",
),
),
]
92 changes: 90 additions & 2 deletions nautobot_firewall_models/tests/test_api.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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."""
Expand Down
1 change: 1 addition & 0 deletions nautobot_firewall_models/tests/test_ui_views.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""Unit tests for views."""

# flake8: noqa: F403,405
# pylint: disable=invalid-name
# pylint: disable=duplicate-code
Expand Down