modules: add rosedb module test

tests-extra/tests/modules/rosedb/test.py

+#!/usr/bin/env python3
+
+''' Check 'rosedb' query module functionality. '''
+
+import os.path
+from dnstest.test import Test
+from dnstest.utils import *
+from dnstest.module import ModRosedb
+
+t = Test()
+
+ModRosedb.check()
+
+# Initialize server configuration.
+zone = t.zone("example.com.")
+knot = t.server("knot")
+t.link(zone, knot)
+
+# Attach rosedb.
+module = ModRosedb(os.path.join(knot.dir, "rosedb"))
+knot.add_module(None, module)
+
+t.start()
+
+# Check before rosedb applied.
+resp = knot.dig("mail.example.com", "A")
+resp.check(rcode="NOERROR", rdata="192.0.2.3", ttl=3600, flags="AA")
+
+# Set rosedb records.
+module.add_record("mail.example.com",  "A",    "1000", "127.0.0.1")
+module.add_record("mail6.example.com", "AAAA", "1000", "::1")
+knot.reload()
+
+# Check if zone record is overridden with rosedb.
+resp = knot.dig("mail.example.com", "A")
+resp.check(rcode="NOERROR", rdata="127.0.0.1", ttl=1000, noflags="AA")
+
+# Check for subdomain match.
+resp = knot.dig("sub.sub.mail.example.com", "A")
+resp.check(rcode="NOERROR", rdata="127.0.0.1", ttl=1000, noflags="AA")
+
+# Check for new record.
+resp = knot.dig("mail6.example.com", "AAAA")
+resp.check(rcode="NOERROR", rdata="::1", ttl=1000, noflags="AA")
+
+# Check for new record with bad type (NODATA).
+resp = knot.dig("mail6.example.com", "A")
+resp.check(rcode="NOERROR", noflags="AA")
+compare(resp.count(), 0, "A count")
+
+# Add autority information.
+module.add_record("example.net",     "SOA", "1", "ns1 host 1 3600 60 3600 3600")
+module.add_record("example.net",     "NS",  "2", "ns1.example.net")
+module.add_record("ns1.example.net", "A",   "3", "127.0.0.2")
+knot.reload()
+
+# Check for authoritative answer.
+resp = knot.dig("example.net", "NS")
+resp.check(rcode="NOERROR", rdata="ns1.example.net.", ttl=2, flags="AA")
+resp.check_count(1, rtype="SOA", section="authority")
+
+# Check for NXDOMAIN.
+resp = knot.dig("example.net", "MX")
+resp.check(rcode="NXDOMAIN", flags="AA")
+resp.check_count(1, rtype="SOA", section="authority")
+
+t.end()

tests-extra/tools/dnstest/module.py

 #!/usr/bin/env python3
 
+import os
 import re
-from subprocess import Popen, PIPE, check_call, CalledProcessError
+from subprocess import Popen, PIPE, check_call
 from dnstest.utils import *
 import dnstest.config
 import dnstest.params as params
-import dnstest.server
 
 class KnotModule(object):
     '''Query module configuration'''
@@ -152,3 +152,36 @@ class ModWhoami(KnotModule):
         conf.end()
 
         return conf
+
+class ModRosedb(KnotModule):
+    '''Rosedb module'''
+
+    src_name = "rosedb_load"
+    conf_name = "mod-rosedb"
+
+    def __init__(self, dbdir):
+        super().__init__()
+        self.dbdir = dbdir
+
+    def get_conf(self, conf=None):
+        if not conf:
+            conf = dnstest.config.KnotConf()
+
+        conf.begin(self.conf_name)
+        conf.id_item("id", self.conf_id)
+        conf.item_str("dbdir", "%s" % (self.dbdir))
+        conf.end()
+
+        return conf
+
+    def add_record(self, owner, rtype, ttl, rdata, code="-", target="-"):
+        prepare_dir(self.dbdir)
+        try:
+            check_call([params.rosedb_tool, self.dbdir, 'add', owner, rtype,
+                        ttl, rdata, code, target],
+                       stdout=open(os.path.join(params.out_dir, "rosedb-tool.out"), mode="a"),
+                       stderr=open(os.path.join(params.out_dir, "rosedb-tool.err"), mode="a"))
+        except:
+            set_err("ROSEDB_TOOL")
+            detail_log("!Failed to add a record into rosedb '%s'" % self.dbdir)
+            detail_log(SEP)

tests-extra/tools/dnstest/params.py

@@ -60,10 +60,8 @@ keymgr_bin = get_binary("KNOT_TEST_KEYMGR", repo_binary("src/keymgr"))
 bind_bin = get_binary("KNOT_TEST_BIND", "named")
 # KNOT_TEST_BINDC - Bind control binary.
 bind_ctl = get_binary("KNOT_TEST_BINDC", "rndc")
-# KNOT_TEST_NSD - Nsd binary.
-#nsd_bin = get_binary("KNOT_TEST_NSD", "nsd")
-# KNOT_TEST_NSDC - Nsd control binary.
-#nsd_ctl = get_binary("KNOT_TEST_NSDC", "nsdc")
+# KNOT_TEST_ROSEDB_TOOL - Rosedb tool binary.
+rosedb_tool = get_binary("KNOT_TEST_ROSEDB_TOOL", repo_binary("src/rosedb_tool"))
 
 # KNOT_TEST_OUTS_DIR - working directories location.
 outs_dir = get_param("KNOT_TEST_OUTS_DIR", "/tmp")