Issue 206 cname check

mreg/api/v1/tests/tests.py

@@ -439,6 +439,14 @@ def test_hosts_post_409_conflict_name(self):
         """"Posting a new host with a name already in use should return 409"""
         self.assert_post_and_409('/hosts/', self.post_data_name)
 
+    def test_hosts_post_409_conflict_cname(self):
+        """"Posting a new host with a name already in use as cname should return 409"""
+        cname_data = {'name': 'host3.example.org', 'host': self.host_one.id}
+        self.assert_post_and_201('/cnames/', cname_data)
+        conflicting_host_data = {'name': 'host3.example.org', 'ipaddress': '127.0.0.3',
+                                 'contact': '[email protected]'}
+        self.assert_post_and_409('/hosts/', conflicting_host_data)
+
     def test_hosts_patch_204_no_content(self):
         """Patching an existing and valid entry should return 204 and Location"""
         response = self.assert_patch_and_204('/hosts/%s' % self.host_one.name, self.patch_data)
@@ -468,6 +476,13 @@ def test_hosts_patch_409_conflict_name(self):
         """Patching an entry with a name that already exists should return 409"""
         self.assert_patch_and_409('/hosts/%s' % self.host_one.name, {'name': self.host_two.name})
 
+    def test_hosts_patch_409_conflict_cname(self):
+        """"Patching an entry host with a name already in use as cname should return 409"""
+        cname_data = {'name': 'host3.example.org', 'host': self.host_one.id}
+        self.assert_post_and_201('/cnames/', cname_data)
+        conflicting_host_data = {'name': 'host3.example.org'}
+        self.assert_patch_and_409('/hosts/%s' % self.host_one.name, conflicting_host_data)
+
 
 class APIHostsTestCaseAsAdminuser(APIHostsTestCase):
     """Same tests as in APIHostsTestCase, only test as admin and not super"""
@@ -678,6 +693,27 @@ def test_naptr_zone_autoupdate_delete(self):
         self.zone.refresh_from_db()
         self.assertTrue(self.zone.updated)
 
+    def test_naptr_post_409_conflict_cname(self):
+        cname_data = {'name': 'replacement.example.org', "host": self.host.id}
+        self.assert_post_and_201('/cnames/', cname_data)
+        conflicting_naptr_data = {'host': self.host.id,
+                'preference': 10,
+                'order': 20,
+                'flag': 'a',
+                'service': 'SERVICE',
+                'regex': r'1(.*@example.org)',
+                'replacement': 'replacement.example.org'
+                }
+        self.assert_post_and_409('/naptrs/', conflicting_naptr_data)
+
+    def test_naptr_patch_409_conflict_cname(self):
+        cname_data = {'name': 'replacement1.example.org', "host": self.host.id}
+        self.assert_post_and_201('/cnames/', cname_data)
+        self.test_naptr_post()
+        naptrs = self.assert_get("/naptrs/").json()['results']
+        conflicting_naptr_data = {'replacement': 'replacement1.example.org'}
+        self.assert_patch_and_409("/naptrs/{}".format(naptrs[0]['id']), conflicting_naptr_data)
+
 
 class APIPtrOverrideTestcase(MregAPITestCase):
     """Test PtrOverride records."""

mreg/api/v1/views.py

@@ -43,6 +43,15 @@
 from .zonefile import ZoneFile
 
 
+def cname_conflict(cname):
+    try:
+        Cname.objects.get(name=cname)
+        # No exception, means such Cname exists
+        return True
+    except Cname.DoesNotExist:
+        return False
+
+
 # These filtersets are used for applying generic filtering to all objects.
 class CnameFilterSet(ModelFilterSet):
     class Meta:
@@ -309,7 +318,10 @@ def post(self, request, *args, **kwargs):
             if self.queryset.filter(name=request.data["name"]).exists():
                 content = {'ERROR': 'name already in use'}
                 return Response(content, status=status.HTTP_409_CONFLICT)
-
+            elif cname_conflict(request.data["name"]):
+                content = {'ERROR': 'name already in use as cname'}
+                return Response(content, status=status.HTTP_409_CONFLICT)
+
         hostdata = request.data.copy()
 
         if 'ipaddress' in hostdata:
@@ -358,6 +370,9 @@ def patch(self, request, *args, **kwargs):
             if self.get_queryset().filter(name=request.data["name"]).exists():
                 content = {'ERROR': 'name already in use'}
                 return Response(content, status=status.HTTP_409_CONFLICT)
+            elif cname_conflict(request.data["name"]):
+                content = {'ERROR': 'name already in use as cname'}
+                return Response(content, status=status.HTTP_409_CONFLICT)
 
         return super().patch(request, *args, **kwargs)
 
@@ -443,6 +458,14 @@ def get_queryset(self):
         qs = super().get_queryset()
         return NaptrFilterSet(data=self.request.GET, queryset=qs).filter()
 
+    def post(self, request, *args, **kwargs):
+        if "replacement" in request.data:
+            if cname_conflict(request.data["replacement"]):
+                content = {'ERROR': 'name already in use as cname'}
+                return Response(content, status=status.HTTP_409_CONFLICT)
+
+        return super().post(request, *args, **kwargs)
+
 
 class NaptrDetail(HostPermissionsUpdateDestroy,
                   MregRetrieveUpdateDestroyAPIView):
@@ -460,6 +483,14 @@ class NaptrDetail(HostPermissionsUpdateDestroy,
     queryset = Naptr.objects.all()
     serializer_class = NaptrSerializer
 
+    def patch(self, request, *args, **kwargs):
+        if "replacement" in request.data:
+            if cname_conflict(request.data["replacement"]):
+                content = {'ERROR': 'name already in use as cname'}
+                return Response(content, status=status.HTTP_409_CONFLICT)
+
+        return super().patch(request, *args, **kwargs)
+
 
 class NameServerList(HostPermissionsListCreateAPIView):
     """