Merge pull request #5 from SaaShup/fix_dnsrecord_unicity_logic

🗃️ 👔 Change DNS record unicity logic

netbox_cloudflare_plugin/__init__.py

@@ -8,7 +8,7 @@ class NetBoxCloudflareConfig(PluginConfig):
     name = "netbox_cloudflare_plugin"
     verbose_name = " NetBox Cloudflare Plugin"
     description = "Manage Cloudflare"
-    version = "0.2.0"
+    version = "0.3.0"
     base_url = "cloudflare"
     min_version = "4.0.0"
     author= "Vincent Simonin <[email protected]>"

netbox_cloudflare_plugin/migrations/0003_remove_dnsrecord_netbox_cloudflare_plugin_dnsrecord_unique_zone_name_type_content_and_more.py

@@ -0,0 +1,34 @@
+# pylint: disable=C0103
+"""Migration file"""
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    """Migration file"""
+
+    dependencies = [
+        ("netbox_cloudflare_plugin", "0002_alter_dnsrecord_record_id"),
+    ]
+
+    operations = [
+        migrations.RemoveConstraint(
+            model_name="dnsrecord",
+            name="netbox_cloudflare_plugin_dnsrecord_unique_zone_name_type_content",
+        ),
+        migrations.AddConstraint(
+            model_name="dnsrecord",
+            constraint=models.UniqueConstraint(
+                fields=("zone", "name", "type", "content"),
+                name="netbox_cloudflare_plugin_dnsrecord_unique_zone_name_type_content",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="dnsrecord",
+            constraint=models.UniqueConstraint(
+                condition=models.Q(("proxied", True)),
+                fields=("zone", "name", "type"),
+                name="netbox_cloudflare_plugin_dnsrecord_unique_zone_name_type",
+            ),
+        ),
+    ]

netbox_cloudflare_plugin/models.py

@@ -1,6 +1,7 @@
 """Cloudflare Plugin Models definitions"""
 
 from django.db import models
+from django.db.models import Q
 from django.urls import reverse
 from django.core.validators import (
     MinLengthValidator,
@@ -125,8 +126,13 @@ class Meta:
         )
         constraints = (
             models.UniqueConstraint(
-                fields=["zone", "name"],
-                name="%(app_label)s_%(class)s_unique_zone_name_type_content",
+                fields=["zone", "name", "type", "content"],
+                name="%(app_label)s_%(class)s_unique_zone_name_type_content"
+            ),
+            models.UniqueConstraint(
+                fields=["zone", "name", "type"],
+                name="%(app_label)s_%(class)s_unique_zone_name_type",
+                condition=Q(proxied=True)
             ),
         )
 

netbox_cloudflare_plugin/signals.py

@@ -38,6 +38,14 @@ def create_dnsrecord(instance, **_kwargs):
         if not instance.name.endswith("." + instance.zone.zone_name):
             instance.name = instance.name + "." + instance.zone.zone_name
 
+        if DnsRecord.objects.filter(
+            zone_id=instance.zone_id,
+            name=instance.name,
+            type=instance.type,
+            proxied=(not instance.proxied),
+        ).exists():
+            raise AbortRequest("Unable to create DNS Record on Cloudflare")
+
         client = CloudflareDnsClient(
             instance.zone,
             settings.PLUGINS_CONFIG["netbox_cloudflare_plugin"]["cloudflare_base_url"],

netbox_cloudflare_plugin/tests/dnsrecord/test_dnsrecord_api.py

@@ -132,3 +132,183 @@ def test_that_record_name_is_set_according_convention(self):
         content = response.json()
 
         self.assertEqual(content[0]["name"], "test8.test2.cloud")
+
+    def test_that_record_must_be_unique(self):
+        """Test that record must be unique"""
+
+        # Assign model-level permission
+        obj_perm = ObjectPermission(
+            name="Test permission", actions=["add", "change", "view"]
+        )
+        obj_perm.save()
+        # pylint: disable=E1101
+        obj_perm.users.add(self.user)
+        # pylint: disable=E1101
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(DnsRecord))
+
+        zone3 = ZoneAccount.objects.create(
+            zone_name="test3.cloud", zone_id="1003", token="token3"
+        )
+
+        response = self.client.post(
+            reverse(f"plugins-api:{self._get_view_namespace()}:dnsrecord-list"),
+            [
+                {
+                    "zone": zone3.pk,
+                    "name": "test9",
+                    "type": DnsRecord.A,
+                    "content": "10.10.10.16",
+                    "proxied": True,
+                },
+            ],
+            format="json",
+            **self.header,
+        )
+
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+
+        content = response.json()
+
+        self.assertEqual(content[0]["name"], "test9.test3.cloud")
+
+        response = self.client.post(
+            reverse(f"plugins-api:{self._get_view_namespace()}:dnsrecord-list"),
+            [
+                {
+                    "zone": zone3.pk,
+                    "name": "test9",
+                    "type": DnsRecord.A,
+                    "content": "10.10.10.17",
+                    "proxied": True,
+                },
+            ],
+            format="json",
+            **self.header,
+        )
+
+        self.assertHttpStatus(response, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+        content = response.json()
+
+        self.assertEqual(content["exception"], "IntegrityError")
+
+    def test_that_record_can_be_use_for_round_robin(self):
+        """Test that record must be unique"""
+
+        # Assign model-level permission
+        obj_perm = ObjectPermission(
+            name="Test permission", actions=["add", "change", "view"]
+        )
+        obj_perm.save()
+        # pylint: disable=E1101
+        obj_perm.users.add(self.user)
+        # pylint: disable=E1101
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(DnsRecord))
+
+        zone4 = ZoneAccount.objects.create(
+            zone_name="test4.cloud", zone_id="1004", token="token4"
+        )
+
+        response = self.client.post(
+            reverse(f"plugins-api:{self._get_view_namespace()}:dnsrecord-list"),
+            [
+                {
+                    "zone": zone4.pk,
+                    "name": "test10",
+                    "type": DnsRecord.A,
+                    "content": "10.10.10.18",
+                    "proxied": False,
+                },
+            ],
+            format="json",
+            **self.header,
+        )
+
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+
+        content = response.json()
+
+        self.assertEqual(content[0]["name"], "test10.test4.cloud")
+
+        response = self.client.post(
+            reverse(f"plugins-api:{self._get_view_namespace()}:dnsrecord-list"),
+            [
+                {
+                    "zone": zone4.pk,
+                    "name": "test10",
+                    "type": DnsRecord.A,
+                    "content": "10.10.10.19",
+                    "proxied": False,
+                },
+            ],
+            format="json",
+            **self.header,
+        )
+
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+
+        content = response.json()
+
+        self.assertEqual(content[0]["name"], "test10.test4.cloud")
+
+    def test_that_record_must_be_unique_with_not_proxied(self):
+        """Test that record must be unique with not proxied"""
+
+        # Assign model-level permission
+        obj_perm = ObjectPermission(
+            name="Test permission", actions=["add", "change", "view"]
+        )
+        obj_perm.save()
+        # pylint: disable=E1101
+        obj_perm.users.add(self.user)
+        # pylint: disable=E1101
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(DnsRecord))
+
+        zone5 = ZoneAccount.objects.create(
+            zone_name="test5.cloud", zone_id="1005", token="token5"
+        )
+
+        response = self.client.post(
+            reverse(f"plugins-api:{self._get_view_namespace()}:dnsrecord-list"),
+            [
+                {
+                    "zone": zone5.pk,
+                    "name": "test11",
+                    "type": DnsRecord.A,
+                    "content": "10.10.10.20",
+                    "proxied": False,
+                },
+            ],
+            format="json",
+            **self.header,
+        )
+
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+
+        content = response.json()
+
+        self.assertEqual(content[0]["name"], "test11.test5.cloud")
+
+        response = self.client.post(
+            reverse(f"plugins-api:{self._get_view_namespace()}:dnsrecord-list"),
+            [
+                {
+                    "zone": zone5.pk,
+                    "name": "test11",
+                    "type": DnsRecord.A,
+                    "content": "10.10.10.21",
+                    "proxied": True,
+                },
+            ],
+            format="json",
+            **self.header,
+        )
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+
+        content = response.json()
+
+        self.assertEqual(content["detail"], "Unable to create DNS Record on Cloudflare")