[1455] implement check for RFC2136 section 2.4.1

src/lib/python/isc/ddns/session.py

@@ -123,7 +123,7 @@ class UpdateSession:
         try:
             datasrc_client, zname, zclass = self.__get_update_zone()
             # conceptual code that would follow
-            self.__check_prerequisites()
+            self.__check_prerequisites(datasrc_client)
             # self.__check_update_acl()
             # self.__do_update()
             # self.__make_response(Rcode.NOERROR())
@@ -188,10 +188,17 @@ class UpdateSession:
         self.__message.clear_section(SECTION_ZONE)
         self.__message.set_rcode(rcode)
 
-    def __check_prerequisite_rrset_exists(self):
-        pass
+    def __check_prerequisite_rrset_exists(self, datasrc_client, rrset):
+        '''Check whether an rrset with the given name and type exists. Class,
+           TTL, and Rdata (if any) of the given RRset are ignored.
+           RFC2136 Section 2.4.1.
+        '''
+        _, finder = datasrc_client.find_zone(rrset.get_name())
+        result, _, _ = finder.find(rrset.get_name(), rrset.get_type(),
+                                   finder.NO_WILDCARD | finder.FIND_GLUE_OK)
+        return result == finder.SUCCESS
 
-    def __check_prerequisite_rrset_exists(self, foo):
+    def __check_prerequisite_rrset_exists_value(self, rrset):
         pass
 
     def __check_prerequisite_rrset_does_not_exist(self):
@@ -203,5 +210,5 @@ class UpdateSession:
     def __check_prerequisite_name_not_in_use(self):
         pass
 
-    def __check_prerequisites(self):
+    def __check_prerequisites(self, datasrc_client):
         pass

src/lib/python/isc/ddns/tests/session_tests.py

@@ -148,8 +148,84 @@ class SessionTest(unittest.TestCase):
         # zone class doesn't match
         self.check_notauth(Name('example.org'), RRClass.CH())
 
+    def __check_prerequisite_exists(self, expected, client, rrset):
+        '''helper method for test_check_prerequisite_exists. This function
+           does nothing but call __check_prerequisite_exists, and checks
+           the result value.
+           expected is either True or False.
+           (since the check_prerequisite is a 'private' method, and calling
+           it is a bit ugly, this helper method helps keeping the actual
+           test code cleaner).
+        '''
+        if expected:
+            self.assertTrue(
+                self.__session._UpdateSession__check_prerequisite_rrset_exists(
+                                client, rrset))
+        else:
+            self.assertFalse(
+                self.__session._UpdateSession__check_prerequisite_rrset_exists(
+                                client, rrset))
+
     def test_check_prerequisite_exists(self):
-        pass
+        # Basic existence checks
+        # www.example.org should have an A, but not an MX
+        rrset = isc.dns.RRset(isc.dns.Name("www.example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.A(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(True, self.__datasrc_client, rrset)
+        rrset = isc.dns.RRset(isc.dns.Name("www.example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.MX(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(False, self.__datasrc_client, rrset)
+
+        # example.org should have an MX, but not an A
+        rrset = isc.dns.RRset(isc.dns.Name("example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.MX(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(True, self.__datasrc_client, rrset)
+        rrset = isc.dns.RRset(isc.dns.Name("example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.A(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(False, self.__datasrc_client, rrset)
+
+        # Wildcard expansion should not be applied, but literal matches
+        # should work
+        rrset = isc.dns.RRset(isc.dns.Name("foo.wildcard.example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.A(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(False, self.__datasrc_client, rrset)
+
+        rrset = isc.dns.RRset(isc.dns.Name("*.wildcard.example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.A(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(True, self.__datasrc_client, rrset)
+
+        # Likewise, CNAME directly should match, but what it points to should not
+        rrset = isc.dns.RRset(isc.dns.Name("cname.example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.A(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(False, self.__datasrc_client, rrset)
+
+        rrset = isc.dns.RRset(isc.dns.Name("cname.example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.CNAME(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(True, self.__datasrc_client, rrset)
+
+        # And also make sure a delegation (itself) is not treated as existing data
+        rrset = isc.dns.RRset(isc.dns.Name("foo.sub.example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.A(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(False, self.__datasrc_client, rrset)
+        # But the delegation data itself should match
+        rrset = isc.dns.RRset(isc.dns.Name("sub.example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.NS(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(True, self.__datasrc_client, rrset)
+        # As should glue
+        rrset = isc.dns.RRset(isc.dns.Name("ns.sub.example.org"),
+                              isc.dns.RRClass.IN(), isc.dns.RRType.A(),
+                              isc.dns.RRTTL(0))
+        self.__check_prerequisite_exists(True, self.__datasrc_client, rrset)
 
     def test_check_prerequisite_exists_value(self):
         pass