regression-tests.auth-py/test_LuaRecords.py

   1 #!/usr/bin/env python
   2 import unittest
   3 import requests
   4 import threading
   5 import dns
   6 import time
   7 import clientsubnetoption
   8
   9 from authtests import AuthTest
  10
  11 from http.server import BaseHTTPRequestHandler, HTTPServer
  12
  13 webserver = None
  14
  15 class FakeHTTPServer(BaseHTTPRequestHandler):
  16     def _set_headers(self, response_code=200):
  17         self.send_response(response_code)
  18         self.send_header('Content-type', 'text/html')
  19         self.end_headers()
  20
  21     def do_GET(self):
  22         if self.path == '/404':
  23             self._set_headers(404)
  24             self.wfile.write(bytes('this page does not exist', 'utf-8'))
  25             return
  26
  27         self._set_headers()
  28         if self.path == '/ping.json':
  29             self.wfile.write(bytes('{"ping":"pong"}', 'utf-8'))
  30         else:
  31             self.wfile.write(bytes("<html><body><h1>hi!</h1><h2>Programming in Lua !</h2></body></html>", "utf-8"))
  32
  33     def log_message(self, format, *args):
  34         return
  35
  36     def do_HEAD(self):
  37         self._set_headers()
  38
  39 class TestLuaRecords(AuthTest):
  40     _config_template = """
  41 geoip-database-files=../modules/geoipbackend/regression-tests/GeoLiteCity.mmdb
  42 edns-subnet-processing=yes
  43 launch=bind geoip
  44 any-to-tcp=no
  45 enable-lua-records
  46 lua-health-checks-interval=1
  47 """
  48
  49     _zones = {
  50         'example.org': """
  51 example.org.                 3600 IN SOA  {soa}
  52 example.org.                 3600 IN NS   ns1.example.org.
  53 example.org.                 3600 IN NS   ns2.example.org.
  54 ns1.example.org.             3600 IN A    {prefix}.10
  55 ns2.example.org.             3600 IN A    {prefix}.11
  56
  57 web1.example.org.            3600 IN A    {prefix}.101
  58 web2.example.org.            3600 IN A    {prefix}.102
  59 web3.example.org.            3600 IN A    {prefix}.103
  60
  61 all.ifportup                 3600 IN LUA  A     "ifportup(8080, {{'{prefix}.101', '{prefix}.102'}})"
  62 some.ifportup                3600 IN LUA  A     "ifportup(8080, {{'192.168.42.21', '{prefix}.102'}})"
  63 multi.ifportup               3600 IN LUA  A     "ifportup(8080, {{ {{'192.168.42.23'}}, {{'192.168.42.21', '{prefix}.102'}}, {{'{prefix}.101'}} }})"
  64 none.ifportup                3600 IN LUA  A     "ifportup(8080, {{'192.168.42.21', '192.168.21.42'}})"
  65 all.noneup.ifportup          3600 IN LUA  A     "ifportup(8080, {{'192.168.42.21', '192.168.21.42'}}, {{ backupSelector='all' }})"
  66
  67 hashed.example.org.          3600 IN LUA  A     "pickhashed({{ '1.2.3.4', '4.3.2.1' }})"
  68 hashed-v6.example.org.       3600 IN LUA  AAAA  "pickhashed({{ '2001:db8:a0b:12f0::1', 'fe80::2a1:9bff:fe9b:f268' }})"
  69 hashed-txt.example.org.      3600 IN LUA  TXT   "pickhashed({{ 'bob', 'alice' }})"
  70 whashed.example.org.         3600 IN LUA  A     "pickwhashed({{ {{15, '1.2.3.4'}}, {{42, '4.3.2.1'}} }})"
  71 whashed-txt.example.org.     3600 IN LUA  TXT   "pickwhashed({{ {{15, 'bob'}}, {{42, 'alice'}} }})"
  72 rand.example.org.            3600 IN LUA  A     "pickrandom({{'{prefix}.101', '{prefix}.102'}})"
  73 rand-txt.example.org.        3600 IN LUA  TXT   "pickrandom({{ 'bob', 'alice' }})"
  74 randn-txt.example.org.       3600 IN LUA  TXT   "pickrandomsample( 2, {{ 'bob', 'alice', 'john' }} )"
  75 v6-bogus.rand.example.org.   3600 IN LUA  AAAA  "pickrandom({{'{prefix}.101', '{prefix}.102'}})"
  76 v6.rand.example.org.         3600 IN LUA  AAAA  "pickrandom({{ '2001:db8:a0b:12f0::1', 'fe80::2a1:9bff:fe9b:f268' }})"
  77 closest.geo                  3600 IN LUA  A     "pickclosest({{ '1.1.1.2', '1.2.3.4' }})"
  78 empty.rand.example.org.      3600 IN LUA  A     "pickrandom()"
  79 timeout.example.org.         3600 IN LUA  A     "; local i = 0 ;  while i < 1000 do pickrandom() ; i = i + 1 end return '1.2.3.4'"
  80 wrand.example.org.           3600 IN LUA  A     "pickwrandom({{ {{30, '{prefix}.102'}}, {{15, '{prefix}.103'}} }})"
  81 wrand-txt.example.org.       3600 IN LUA  TXT   "pickwrandom({{ {{30, 'bob'}}, {{15, 'alice'}} }})"
  82 all.example.org.             3600 IN LUA  A     "all({{'1.2.3.4','4.3.2.1'}})"
  83
  84 config    IN    LUA    LUA ("settings={{stringmatch='Programming in Lua'}} "
  85                             "EUWips={{'{prefix}.101','{prefix}.102'}}      "
  86                             "EUEips={{'192.168.42.101','192.168.42.102'}}  "
  87                             "NLips={{'{prefix}.111', '{prefix}.112'}}      "
  88                             "USAips={{'{prefix}.103', '192.168.42.105'}}   ")
  89
  90 usa          IN    LUA    A   ( ";include('config')                         "
  91                                 "return ifurlup('http://www.lua.org:8080/', "
  92                                 "USAips, settings)                          ")
  93
  94 usa-ext      IN    LUA    A   ( ";include('config')                         "
  95                                 "return ifurlup('http://www.lua.org:8080/', "
  96                                 "{{EUEips, USAips}}, settings)              ")
  97
  98 mix.ifurlup  IN    LUA    A   ("ifurlup('http://www.other.org:8080/ping.json', "
  99                                "{{ '192.168.42.101', '{prefix}.101' }},        "
 100                                "{{ stringmatch='pong' }})                      ")
 101
 102 ifurlextup   IN    LUA    A   "ifurlextup({{{{['192.168.0.1']='http://{prefix}.101:8080/404',['192.168.0.2']='http://{prefix}.102:8080/404'}}, {{['192.168.0.3']='http://{prefix}.101:8080/'}}}})"
 103
 104 nl           IN    LUA    A   ( ";include('config')                                "
 105                                 "return ifportup(8081, NLips) ")
 106 latlon.geo      IN LUA    TXT "latlon()"
 107 continent.geo   IN LUA    TXT ";if(continent('NA')) then return 'true' else return 'false' end"
 108 continent-code.geo   IN LUA    TXT ";return continentCode()"
 109 asnum.geo       IN LUA    TXT ";if(asnum('4242')) then return 'true' else return 'false' end"
 110 country.geo     IN LUA    TXT ";if(country('US')) then return 'true' else return 'false' end"
 111 country-code.geo     IN LUA    TXT ";return countryCode()"
 112 region.geo      IN LUA    TXT ";if(region('CA')) then return 'true' else return 'false' end"
 113 region-code.geo      IN LUA    TXT ";return regionCode()"
 114 latlonloc.geo   IN LUA    TXT "latlonloc()"
 115
 116 true.netmask     IN LUA   TXT   ( ";if(netmask({{ '{prefix}.0/24' }})) "
 117                                   "then return 'true'                  "
 118                                   "else return 'false'             end " )
 119 false.netmask    IN LUA   TXT   ( ";if(netmask({{ '1.2.3.4/8' }}))     "
 120                                   "then return 'true'                  "
 121                                   "else return 'false'             end " )
 122
 123 view             IN    LUA    A          ("view({{                                       "
 124                                           "{{ {{'192.168.0.0/16'}}, {{'192.168.1.54'}}}},"
 125                                           "{{ {{'{prefix}.0/16'}}, {{'{prefix}.54'}}}},  "
 126                                           "{{ {{'0.0.0.0/0'}}, {{'192.0.2.1'}}}}         "
 127                                           " }})                                          " )
 128 txt.view         IN    LUA    TXT        ("view({{                                       "
 129                                           "{{ {{'192.168.0.0/16'}}, {{'txt'}}}},         "
 130                                           "{{ {{'0.0.0.0/0'}}, {{'else'}}}}              "
 131                                           " }})                                          " )
 132 none.view        IN    LUA    A          ("view({{                                     "
 133                                           "{{ {{'192.168.0.0/16'}}, {{'192.168.1.54'}}}},"
 134                                           "{{ {{'1.2.0.0/16'}}, {{'1.2.3.4'}}}},         "
 135                                           " }})                                          " )
 136 *.magic          IN    LUA    A     "closestMagic()"
 137 www-balanced     IN           CNAME 1-1-1-3.17-1-2-4.1-2-3-5.magic.example.org.
 138
 139 any              IN    LUA    A   "'192.0.2.1'"
 140 any              IN           TXT "hello there"
 141
 142 resolve          IN    LUA    A   ";local r=resolve('localhost', 1) local t={{}} for _,v in ipairs(r) do table.insert(t, v:toString()) end return t"
 143
 144 *.createforward  IN    LUA    A     "filterForward(createForward(), newNMG{{'1.0.0.0/8', '64.0.0.0/8'}})"
 145 *.createreverse  IN    LUA    PTR   "createReverse('%5%.example.com', {{['10.10.10.10'] = 'quad10.example.com.'}})"
 146 *.createreverse6 IN    LUA    PTR   "createReverse6('%33%.example.com', {{['2001:db8::1'] = 'example.example.com.'}})"
 147
 148 newcafromraw     IN    LUA    A    "newCAFromRaw('ABCD'):toString()"
 149 newcafromraw     IN    LUA    AAAA "newCAFromRaw('ABCD020340506070'):toString()"
 150
 151 counter          IN    LUA    TXT  ";counter = counter or 0 counter=counter+1 return tostring(counter)"
 152
 153 lookmeup         IN           A  192.0.2.5
 154 dblookup         IN    LUA    A  "dblookup('lookmeup.example.org', 'A')[1]"
 155         """,
 156         'createforward6.example.org': """
 157 createforward6.example.org.                 3600 IN SOA  {soa}
 158 createforward6.example.org.                 3600 IN NS   ns1.example.org.
 159 createforward6.example.org.                 3600 IN NS   ns2.example.org.
 160 *                                                IN    LUA    AAAA  "filterForward(createForward6(), newNMG{{'2000::/3'}}, 'fe80::1')"
 161         """
 162 # the separate createforward6 zone is because some of the code in lua-record.cc insists on working relatively to the zone apex
 163     }
 164     _web_rrsets = []
 165
 166     @classmethod
 167     def startResponders(cls):
 168         global webserver
 169         if webserver: return  # it is already running
 170
 171         webserver = threading.Thread(name='HTTP Listener',
 172                                      target=cls.HTTPResponder,
 173                                      args=[8080]
 174         )
 175         webserver.setDaemon(True)
 176         webserver.start()
 177
 178     @classmethod
 179     def HTTPResponder(cls, port):
 180         server_address = ('', port)
 181         httpd = HTTPServer(server_address, FakeHTTPServer)
 182         httpd.serve_forever()
 183
 184     @classmethod
 185     def setUpClass(cls):
 186
 187         super(TestLuaRecords, cls).setUpClass()
 188
 189         cls._web_rrsets = [dns.rrset.from_text('web1.example.org.', 0, dns.rdataclass.IN, 'A',
 190                                                '{prefix}.101'.format(prefix=cls._PREFIX)),
 191                            dns.rrset.from_text('web2.example.org.', 0, dns.rdataclass.IN, 'A',
 192                                                '{prefix}.102'.format(prefix=cls._PREFIX)),
 193                            dns.rrset.from_text('web3.example.org.', 0, dns.rdataclass.IN, 'A',
 194                                                '{prefix}.103'.format(prefix=cls._PREFIX))
 195         ]
 196
 197     def testPickRandom(self):
 198         """
 199         Basic pickrandom() test with a set of A records
 200         """
 201         expected = [dns.rrset.from_text('rand.example.org.', 0, dns.rdataclass.IN, 'A',
 202                                         '{prefix}.101'.format(prefix=self._PREFIX)),
 203                     dns.rrset.from_text('rand.example.org.', 0, dns.rdataclass.IN, 'A',
 204                                         '{prefix}.102'.format(prefix=self._PREFIX))]
 205         query = dns.message.make_query('rand.example.org', 'A')
 206
 207         res = self.sendUDPQuery(query)
 208         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 209         self.assertAnyRRsetInAnswer(res, expected)
 210
 211     def testPickRandomTxt(self):
 212         """
 213         Basic pickrandom() test with a set of TXT records
 214         """
 215         expected = [dns.rrset.from_text('rand-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'bob'),
 216                     dns.rrset.from_text('rand-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'alice')]
 217         query = dns.message.make_query('rand-txt.example.org', 'TXT')
 218
 219         res = self.sendUDPQuery(query)
 220         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 221         self.assertAnyRRsetInAnswer(res, expected)
 222
 223     def testBogusV6PickRandom(self):
 224         """
 225         Test a bogus AAAA pickrandom() record  with a set of v4 addr
 226         """
 227         query = dns.message.make_query('v6-bogus.rand.example.org', 'AAAA')
 228
 229         res = self.sendUDPQuery(query)
 230         self.assertRcodeEqual(res, dns.rcode.SERVFAIL)
 231
 232     def testV6PickRandom(self):
 233         """
 234         Test pickrandom() AAAA record
 235         """
 236         expected = [dns.rrset.from_text('v6.rand.example.org.', 0, dns.rdataclass.IN, 'AAAA',
 237                                         '2001:db8:a0b:12f0::1'),
 238                     dns.rrset.from_text('v6.rand.example.org.', 0, dns.rdataclass.IN, 'AAAA',
 239                                         'fe80::2a1:9bff:fe9b:f268')]
 240         query = dns.message.make_query('v6.rand.example.org', 'AAAA')
 241
 242         res = self.sendUDPQuery(query)
 243         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 244         self.assertAnyRRsetInAnswer(res, expected)
 245
 246     def testEmptyRandom(self):
 247         """
 248         Basic pickrandom() test with an empty set
 249         """
 250         query = dns.message.make_query('empty.rand.example.org', 'A')
 251
 252         res = self.sendUDPQuery(query)
 253         self.assertRcodeEqual(res, dns.rcode.SERVFAIL)
 254
 255     def testPickRandomSampleTxt(self):
 256         """
 257         Basic pickrandomsample() test with a set of TXT records
 258         """
 259         expected = [dns.rrset.from_text('randn-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'bob', 'alice'),
 260                     dns.rrset.from_text('randn-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'bob', 'john'),
 261                     dns.rrset.from_text('randn-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'alice', 'bob'),
 262                     dns.rrset.from_text('randn-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'alice', 'john'),
 263                     dns.rrset.from_text('randn-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'john', 'bob'),
 264                     dns.rrset.from_text('randn-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'john', 'alice')]
 265         query = dns.message.make_query('randn-txt.example.org', 'TXT')
 266
 267         res = self.sendUDPQuery(query)
 268         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 269         self.assertIn(res.answer[0], expected)
 270
 271     def testWRandom(self):
 272         """
 273         Basic pickwrandom() test with a set of A records
 274         """
 275         expected = [dns.rrset.from_text('wrand.example.org.', 0, dns.rdataclass.IN, 'A',
 276                                         '{prefix}.103'.format(prefix=self._PREFIX)),
 277                     dns.rrset.from_text('wrand.example.org.', 0, dns.rdataclass.IN, 'A',
 278                                         '{prefix}.102'.format(prefix=self._PREFIX))]
 279         query = dns.message.make_query('wrand.example.org', 'A')
 280
 281         res = self.sendUDPQuery(query)
 282         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 283         self.assertAnyRRsetInAnswer(res, expected)
 284
 285     def testWRandomTxt(self):
 286         """
 287         Basic pickwrandom() test with a set of TXT records
 288         """
 289         expected = [dns.rrset.from_text('wrand-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'bob'),
 290                     dns.rrset.from_text('wrand-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'alice')]
 291         query = dns.message.make_query('wrand-txt.example.org', 'TXT')
 292
 293         res = self.sendUDPQuery(query)
 294         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 295         self.assertAnyRRsetInAnswer(res, expected)
 296
 297     def testIfportup(self):
 298         """
 299         Basic ifportup() test
 300         """
 301         query = dns.message.make_query('all.ifportup.example.org', 'A')
 302         expected = [
 303             dns.rrset.from_text('all.ifportup.example.org.', 0, dns.rdataclass.IN, 'A',
 304                                 '{prefix}.101'.format(prefix=self._PREFIX)),
 305             dns.rrset.from_text('all.ifportup.example.org.', 0, dns.rdataclass.IN, 'A',
 306                                 '{prefix}.102'.format(prefix=self._PREFIX))]
 307
 308         res = self.sendUDPQuery(query)
 309         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 310         self.assertAnyRRsetInAnswer(res, expected)
 311
 312     def testIfportupWithSomeDown(self):
 313         """
 314         Basic ifportup() test with some ports DOWN
 315         """
 316         query = dns.message.make_query('some.ifportup.example.org', 'A')
 317         expected = [
 318             dns.rrset.from_text('some.ifportup.example.org.', 0, dns.rdataclass.IN, 'A',
 319                                 '192.168.42.21'),
 320             dns.rrset.from_text('some.ifportup.example.org.', 0, dns.rdataclass.IN, 'A',
 321                                 '{prefix}.102'.format(prefix=self._PREFIX))]
 322
 323         # we first expect any of the IPs as no check has been performed yet
 324         res = self.sendUDPQuery(query)
 325         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 326         self.assertAnyRRsetInAnswer(res, expected)
 327
 328         # the first IP should not be up so only second should be returned
 329         expected = [expected[1]]
 330         res = self.sendUDPQuery(query)
 331         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 332         self.assertAnyRRsetInAnswer(res, expected)
 333
 334     def testIfportupWithSomeDownMultiset(self):
 335         """
 336         Basic ifportup() test with some ports DOWN from multiple sets
 337         """
 338         query = dns.message.make_query('multi.ifportup.example.org', 'A')
 339         expected = [
 340             dns.rrset.from_text('multi.ifportup.example.org.', 0, dns.rdataclass.IN, 'A',
 341                                 '192.168.42.21'),
 342             dns.rrset.from_text('multi.ifportup.example.org.', 0, dns.rdataclass.IN, 'A',
 343                                 '192.168.42.23'),
 344             dns.rrset.from_text('multi.ifportup.example.org.', 0, dns.rdataclass.IN, 'A',
 345                                 '{prefix}.102'.format(prefix=self._PREFIX)),
 346             dns.rrset.from_text('multi.ifportup.example.org.', 0, dns.rdataclass.IN, 'A',
 347                                 '{prefix}.101'.format(prefix=self._PREFIX))
 348         ]
 349
 350         # we first expect any of the IPs as no check has been performed yet
 351         res = self.sendUDPQuery(query)
 352         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 353         self.assertAnyRRsetInAnswer(res, expected)
 354
 355         # An ip is up in 2 sets, but we expect only the one from the middle set
 356         expected = [expected[2]]
 357         res = self.sendUDPQuery(query)
 358         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 359         self.assertAnyRRsetInAnswer(res, expected)
 360
 361     def testIfportupWithAllDown(self):
 362         """
 363         Basic ifportup() test with all ports DOWN
 364         """
 365         query = dns.message.make_query('none.ifportup.example.org', 'A')
 366         expected = [
 367             dns.rrset.from_text('none.ifportup.example.org.', 0, dns.rdataclass.IN, 'A',
 368                                 '192.168.42.21'),
 369             dns.rrset.from_text('none.ifportup.example.org.', 0, dns.rdataclass.IN, 'A',
 370                                 '192.168.21.42'.format(prefix=self._PREFIX))]
 371
 372         # we first expect any of the IPs as no check has been performed yet
 373         res = self.sendUDPQuery(query)
 374         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 375         self.assertAnyRRsetInAnswer(res, expected)
 376
 377         # no port should be up so we expect any
 378         res = self.sendUDPQuery(query)
 379         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 380         self.assertAnyRRsetInAnswer(res, expected)
 381
 382     def testIfportupWithAllDownAndAllBackupSelector(self):
 383         """
 384         Basic ifportup() test with all ports DOWN, fallback to 'all' backup selector
 385         """
 386         name = 'all.noneup.ifportup.example.org.'
 387         query = dns.message.make_query(name, dns.rdatatype.A)
 388         expected = [dns.rrset.from_text(name, 0, dns.rdataclass.IN, dns.rdatatype.A, '192.168.42.21', '192.168.21.42')]
 389
 390         res = self.sendUDPQuery(query)
 391         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 392
 393         time.sleep(3)
 394         res = self.sendUDPQuery(query)
 395         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 396         self.assertEqual(res.answer, expected)
 397
 398     def testIfurlup(self):
 399         """
 400         Basic ifurlup() test
 401         """
 402         reachable = [
 403             '{prefix}.103'.format(prefix=self._PREFIX)
 404         ]
 405         unreachable = ['192.168.42.105']
 406         ips = reachable + unreachable
 407         all_rrs = []
 408         reachable_rrs = []
 409         for ip in ips:
 410             rr = dns.rrset.from_text('usa.example.org.', 0, dns.rdataclass.IN, 'A', ip)
 411             all_rrs.append(rr)
 412             if ip in reachable:
 413                 reachable_rrs.append(rr)
 414
 415         query = dns.message.make_query('usa.example.org', 'A')
 416         res = self.sendUDPQuery(query)
 417         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 418         self.assertAnyRRsetInAnswer(res, all_rrs)
 419
 420         # the timeout in the LUA health checker is 2 second, so we make sure to wait slightly longer here
 421         time.sleep(3)
 422         res = self.sendUDPQuery(query)
 423         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 424         self.assertAnyRRsetInAnswer(res, reachable_rrs)
 425
 426     def testIfurlupMultiSet(self):
 427         """
 428         Basic ifurlup() test with mutiple sets
 429         """
 430         reachable = [
 431             '{prefix}.103'.format(prefix=self._PREFIX)
 432         ]
 433         unreachable = ['192.168.42.101', '192.168.42.102', '192.168.42.105']
 434         ips = reachable + unreachable
 435         all_rrs = []
 436         reachable_rrs = []
 437         for ip in ips:
 438             rr = dns.rrset.from_text('usa-ext.example.org.', 0, dns.rdataclass.IN, 'A', ip)
 439             all_rrs.append(rr)
 440             if ip in reachable:
 441                 reachable_rrs.append(rr)
 442
 443         query = dns.message.make_query('usa-ext.example.org', 'A')
 444         res = self.sendUDPQuery(query)
 445         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 446         self.assertAnyRRsetInAnswer(res, all_rrs)
 447
 448         # the timeout in the LUA health checker is 2 second, so we make sure to wait slightly longer here
 449         time.sleep(3)
 450         res = self.sendUDPQuery(query)
 451         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 452         self.assertAnyRRsetInAnswer(res, reachable_rrs)
 453
 454     def testIfurlextup(self):
 455         expected = [dns.rrset.from_text('ifurlextup.example.org.', 0, dns.rdataclass.IN, dns.rdatatype.A, '192.168.0.3')]
 456
 457         query = dns.message.make_query('ifurlextup.example.org', 'A')
 458         res = self.sendUDPQuery(query)
 459
 460         # wait for health checks to happen
 461         time.sleep(5)
 462
 463         res = self.sendUDPQuery(query)
 464
 465         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 466         self.assertEqual(res.answer, expected)
 467
 468     def testIfurlupSimplified(self):
 469         """
 470         Basic ifurlup() test with the simplified list of ips
 471         Also ensures the correct path is queried
 472         """
 473         reachable = [
 474             '{prefix}.101'.format(prefix=self._PREFIX)
 475         ]
 476         unreachable = ['192.168.42.101']
 477         ips = reachable + unreachable
 478         all_rrs = []
 479         reachable_rrs = []
 480         for ip in ips:
 481             rr = dns.rrset.from_text('mix.ifurlup.example.org.', 0, dns.rdataclass.IN, 'A', ip)
 482             all_rrs.append(rr)
 483             if ip in reachable:
 484                 reachable_rrs.append(rr)
 485
 486         query = dns.message.make_query('mix.ifurlup.example.org', 'A')
 487         res = self.sendUDPQuery(query)
 488         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 489         self.assertAnyRRsetInAnswer(res, all_rrs)
 490
 491         time.sleep(3)
 492         res = self.sendUDPQuery(query)
 493         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 494         self.assertAnyRRsetInAnswer(res, reachable_rrs)
 495
 496     def testLatlon(self):
 497         """
 498         Basic latlon() test
 499         """
 500         name = 'latlon.geo.example.org.'
 501         ecso = clientsubnetoption.ClientSubnetOption('1.2.3.0', 24)
 502         query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096, options=[ecso])
 503         expected = dns.rrset.from_text(name, 0,
 504                                        dns.rdataclass.IN, 'TXT',
 505                                        '"47.913000 -122.304200"')
 506
 507         res = self.sendUDPQuery(query)
 508         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 509         self.assertRRsetInAnswer(res, expected)
 510
 511     def testLatlonloc(self):
 512         """
 513         Basic latlonloc() test
 514         """
 515         name = 'latlonloc.geo.example.org.'
 516         expected = dns.rrset.from_text(name, 0,dns.rdataclass.IN, 'TXT',
 517                                        '"47 54 46.8 N 122 18 15.12 W 0.00m 1.00m 10000.00m 10.00m"')
 518         ecso = clientsubnetoption.ClientSubnetOption('1.2.3.0', 24)
 519         query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096, options=[ecso])
 520
 521         res = self.sendUDPQuery(query)
 522         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 523         self.assertRRsetInAnswer(res, expected)
 524
 525     def testWildcardError(self):
 526         """
 527         Ensure errors coming from LUA wildcards are reported
 528         """
 529         query = dns.message.make_query('failure.magic.example.org', 'A')
 530
 531         res = self.sendUDPQuery(query)
 532         self.assertRcodeEqual(res, dns.rcode.SERVFAIL)
 533         self.assertAnswerEmpty(res)
 534
 535     def testClosestMagic(self):
 536         """
 537         Basic closestMagic() test
 538         """
 539         name = 'www-balanced.example.org.'
 540         cname = '1-1-1-3.17-1-2-4.1-2-3-5.magic.example.org.'
 541         queries = [
 542             ('1.1.1.0', 24,  '1.1.1.3'),
 543             ('1.2.3.0', 24,  '1.2.3.5'),
 544             ('17.1.0.0', 16, '17.1.2.4')
 545         ]
 546
 547         for (subnet, mask, ip) in queries:
 548             ecso = clientsubnetoption.ClientSubnetOption(subnet, mask)
 549             query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, options=[ecso])
 550
 551             response = dns.message.make_response(query)
 552
 553             response.answer.append(dns.rrset.from_text(name, 0, dns.rdataclass.IN, dns.rdatatype.CNAME, cname))
 554             response.answer.append(dns.rrset.from_text(cname, 0, dns.rdataclass.IN, 'A', ip))
 555
 556             res = self.sendUDPQuery(query)
 557             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 558             self.assertEqual(res.answer, response.answer)
 559
 560     def testAsnum(self):
 561         """
 562         Basic asnum() test
 563         """
 564         queries = [
 565             ('1.1.1.0', 24,  '"true"'),
 566             ('1.2.3.0', 24,  '"false"'),
 567             ('17.1.0.0', 16, '"false"')
 568         ]
 569         name = 'asnum.geo.example.org.'
 570         for (subnet, mask, txt) in queries:
 571             ecso = clientsubnetoption.ClientSubnetOption(subnet, mask)
 572             query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096, options=[ecso])
 573             expected = dns.rrset.from_text(name, 0, dns.rdataclass.IN, 'TXT', txt)
 574
 575             res = self.sendUDPQuery(query)
 576             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 577             self.assertRRsetInAnswer(res, expected)
 578
 579     def testCountry(self):
 580         """
 581         Basic country() test
 582         """
 583         queries = [
 584             ('1.1.1.0', 24,  '"false"'),
 585             ('1.2.3.0', 24,  '"true"'),
 586             ('17.1.0.0', 16, '"false"')
 587         ]
 588         name = 'country.geo.example.org.'
 589         for (subnet, mask, txt) in queries:
 590             ecso = clientsubnetoption.ClientSubnetOption(subnet, mask)
 591             query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096, options=[ecso])
 592             expected = dns.rrset.from_text(name, 0, dns.rdataclass.IN, 'TXT', txt)
 593
 594             res = self.sendUDPQuery(query)
 595             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 596             self.assertRRsetInAnswer(res, expected)
 597
 598     def testCountryCode(self):
 599         """
 600         Basic countryCode() test
 601         """
 602         queries = [
 603             ('1.1.1.0', 24,  '"au"'),
 604             ('1.2.3.0', 24,  '"us"'),
 605             ('17.1.0.0', 16, '"--"')
 606         ]
 607         name = 'country-code.geo.example.org.'
 608         for (subnet, mask, txt) in queries:
 609             ecso = clientsubnetoption.ClientSubnetOption(subnet, mask)
 610             query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096, options=[ecso])
 611             expected = dns.rrset.from_text(name, 0, dns.rdataclass.IN, 'TXT', txt)
 612
 613             res = self.sendUDPQuery(query)
 614             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 615             self.assertRRsetInAnswer(res, expected)
 616
 617     def testRegion(self):
 618         """
 619         Basic region() test
 620         """
 621         queries = [
 622             ('1.1.1.0', 24,  '"false"'),
 623             ('1.2.3.0', 24,  '"true"'),
 624             ('17.1.0.0', 16, '"false"')
 625         ]
 626         name = 'region.geo.example.org.'
 627         for (subnet, mask, txt) in queries:
 628             ecso = clientsubnetoption.ClientSubnetOption(subnet, mask)
 629             query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096, options=[ecso])
 630             expected = dns.rrset.from_text(name, 0, dns.rdataclass.IN, 'TXT', txt)
 631
 632             res = self.sendUDPQuery(query)
 633             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 634             self.assertRRsetInAnswer(res, expected)
 635
 636     def testRegionCode(self):
 637         """
 638         Basic regionCode() test
 639         """
 640         queries = [
 641             ('1.1.1.0', 24,  '"--"'),
 642             ('1.2.3.0', 24,  '"ca"'),
 643             ('17.1.0.0', 16, '"--"')
 644         ]
 645         name = 'region-code.geo.example.org.'
 646         for (subnet, mask, txt) in queries:
 647             ecso = clientsubnetoption.ClientSubnetOption(subnet, mask)
 648             query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096, options=[ecso])
 649             expected = dns.rrset.from_text(name, 0, dns.rdataclass.IN, 'TXT', txt)
 650
 651             res = self.sendUDPQuery(query)
 652             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 653             self.assertRRsetInAnswer(res, expected)
 654
 655     def testContinent(self):
 656         """
 657         Basic continent() test
 658         """
 659         queries = [
 660             ('1.1.1.0', 24,  '"false"'),
 661             ('1.2.3.0', 24,  '"true"'),
 662             ('17.1.0.0', 16, '"false"')
 663         ]
 664         name = 'continent.geo.example.org.'
 665         for (subnet, mask, txt) in queries:
 666             ecso = clientsubnetoption.ClientSubnetOption(subnet, mask)
 667             query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096, options=[ecso])
 668             expected = dns.rrset.from_text(name, 0, dns.rdataclass.IN, 'TXT', txt)
 669
 670             res = self.sendUDPQuery(query)
 671             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 672             self.assertRRsetInAnswer(res, expected)
 673
 674     def testContinentCode(self):
 675         """
 676         Basic continentCode() test
 677         """
 678         queries = [
 679             ('1.1.1.0', 24,  '"oc"'),
 680             ('1.2.3.0', 24,  '"na"'),
 681             ('17.1.0.0', 16, '"--"')
 682         ]
 683         name = 'continent-code.geo.example.org.'
 684         for (subnet, mask, txt) in queries:
 685             ecso = clientsubnetoption.ClientSubnetOption(subnet, mask)
 686             query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096, options=[ecso])
 687             expected = dns.rrset.from_text(name, 0, dns.rdataclass.IN, 'TXT', txt)
 688
 689             res = self.sendUDPQuery(query)
 690             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 691             self.assertRRsetInAnswer(res, expected)
 692
 693     def testClosest(self):
 694         """
 695         Basic pickclosest() test
 696         """
 697         queries = [
 698             ('1.1.1.0', 24,  '1.1.1.2'),
 699             ('1.2.3.0', 24,  '1.2.3.4'),
 700             ('17.1.0.0', 16, '1.1.1.2')
 701         ]
 702         name = 'closest.geo.example.org.'
 703         for (subnet, mask, ip) in queries:
 704             ecso = clientsubnetoption.ClientSubnetOption(subnet, mask)
 705             query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, options=[ecso])
 706             expected = dns.rrset.from_text(name, 0, dns.rdataclass.IN, 'A', ip)
 707
 708             res = self.sendUDPQuery(query)
 709             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 710             self.assertRRsetInAnswer(res, expected)
 711
 712     def testAll(self):
 713         """
 714         Basic all() test
 715         """
 716         expected = [dns.rrset.from_text('all.example.org.', 0, dns.rdataclass.IN, dns.rdatatype.A, '1.2.3.4', '4.3.2.1')]
 717         query = dns.message.make_query('all.example.org.', 'A')
 718
 719         res = self.sendUDPQuery(query)
 720         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 721         self.assertEqual(res.answer, expected)
 722
 723     def testNetmask(self):
 724         """
 725         Basic netmask() test
 726         """
 727         queries = [
 728             {
 729                 'expected': dns.rrset.from_text('true.netmask.example.org.', 0,
 730                                        dns.rdataclass.IN, 'TXT',
 731                                        '"true"'),
 732                 'query': dns.message.make_query('true.netmask.example.org', 'TXT')
 733             },
 734             {
 735                 'expected': dns.rrset.from_text('false.netmask.example.org.', 0,
 736                                        dns.rdataclass.IN, 'TXT',
 737                                        '"false"'),
 738                 'query': dns.message.make_query('false.netmask.example.org', 'TXT')
 739             }
 740         ]
 741         for query in queries :
 742             res = self.sendUDPQuery(query['query'])
 743             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 744             self.assertRRsetInAnswer(res, query['expected'])
 745
 746     def testView(self):
 747         """
 748         Basic view() test
 749         """
 750         queries = [
 751             {
 752                 'expected': dns.rrset.from_text('view.example.org.', 0,
 753                                        dns.rdataclass.IN, 'A',
 754                                        '{prefix}.54'.format(prefix=self._PREFIX)),
 755                 'query': dns.message.make_query('view.example.org', 'A')
 756             },
 757             {
 758                 'expected': dns.rrset.from_text('txt.view.example.org.', 0,
 759                                        dns.rdataclass.IN, 'TXT',
 760                                        '"else"'),
 761                 'query': dns.message.make_query('txt.view.example.org', 'TXT')
 762             }
 763         ]
 764         for query in queries :
 765             res = self.sendUDPQuery(query['query'])
 766             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 767             self.assertRRsetInAnswer(res, query['expected'])
 768
 769     def testViewNoMatch(self):
 770         """
 771         view() test where no netmask match
 772         """
 773         query = dns.message.make_query('none.view.example.org', 'A')
 774
 775         res = self.sendUDPQuery(query)
 776         self.assertRcodeEqual(res, dns.rcode.SERVFAIL)
 777         self.assertAnswerEmpty(res)
 778
 779     def testWHashed(self):
 780         """
 781         Basic pickwhashed() test with a set of A records
 782         As the `bestwho` is hashed, we should always get the same answer
 783         """
 784         expected = [dns.rrset.from_text('whashed.example.org.', 0, dns.rdataclass.IN, 'A', '1.2.3.4'),
 785                     dns.rrset.from_text('whashed.example.org.', 0, dns.rdataclass.IN, 'A', '4.3.2.1')]
 786         query = dns.message.make_query('whashed.example.org', 'A')
 787
 788         first = self.sendUDPQuery(query)
 789         self.assertRcodeEqual(first, dns.rcode.NOERROR)
 790         self.assertAnyRRsetInAnswer(first, expected)
 791         for _ in range(5):
 792             res = self.sendUDPQuery(query)
 793             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 794             self.assertRRsetInAnswer(res, first.answer[0])
 795
 796     def testWHashedTxt(self):
 797         """
 798         Basic pickwhashed() test with a set of TXT records
 799         As the `bestwho` is hashed, we should always get the same answer
 800         """
 801         expected = [dns.rrset.from_text('whashed-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'bob'),
 802                     dns.rrset.from_text('whashed-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'alice')]
 803         query = dns.message.make_query('whashed-txt.example.org', 'TXT')
 804
 805         first = self.sendUDPQuery(query)
 806         self.assertRcodeEqual(first, dns.rcode.NOERROR)
 807         self.assertAnyRRsetInAnswer(first, expected)
 808         for _ in range(5):
 809             res = self.sendUDPQuery(query)
 810             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 811             self.assertRRsetInAnswer(res, first.answer[0])
 812
 813     def testHashed(self):
 814         """
 815         Basic pickhashed() test with a set of A records
 816         As the `bestwho` is hashed, we should always get the same answer
 817         """
 818         expected = [dns.rrset.from_text('hashed.example.org.', 0, dns.rdataclass.IN, 'A', '1.2.3.4'),
 819                     dns.rrset.from_text('hashed.example.org.', 0, dns.rdataclass.IN, 'A', '4.3.2.1')]
 820         query = dns.message.make_query('hashed.example.org', 'A')
 821
 822         first = self.sendUDPQuery(query)
 823         self.assertRcodeEqual(first, dns.rcode.NOERROR)
 824         self.assertAnyRRsetInAnswer(first, expected)
 825         for _ in range(5):
 826             res = self.sendUDPQuery(query)
 827             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 828             self.assertRRsetInAnswer(res, first.answer[0])
 829
 830     def testHashedV6(self):
 831         """
 832         Basic pickhashed() test with a set of AAAA records
 833         As the `bestwho` is hashed, we should always get the same answer
 834         """
 835         expected = [dns.rrset.from_text('hashed-v6.example.org.', 0, dns.rdataclass.IN, 'AAAA', '2001:db8:a0b:12f0::1'),
 836                     dns.rrset.from_text('hashed-v6.example.org.', 0, dns.rdataclass.IN, 'AAAA', 'fe80::2a1:9bff:fe9b:f268')]
 837         query = dns.message.make_query('hashed-v6.example.org', 'AAAA')
 838
 839         first = self.sendUDPQuery(query)
 840         self.assertRcodeEqual(first, dns.rcode.NOERROR)
 841         self.assertAnyRRsetInAnswer(first, expected)
 842         for _ in range(5):
 843             res = self.sendUDPQuery(query)
 844             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 845             self.assertRRsetInAnswer(res, first.answer[0])
 846
 847     def testHashedTXT(self):
 848         """
 849         Basic pickhashed() test with a set of TXT records
 850         As the `bestwho` is hashed, we should always get the same answer
 851         """
 852         expected = [dns.rrset.from_text('hashed-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'bob'),
 853                     dns.rrset.from_text('hashed-txt.example.org.', 0, dns.rdataclass.IN, 'TXT', 'alice')]
 854         query = dns.message.make_query('hashed-txt.example.org', 'TXT')
 855
 856         first = self.sendUDPQuery(query)
 857         self.assertRcodeEqual(first, dns.rcode.NOERROR)
 858         self.assertAnyRRsetInAnswer(first, expected)
 859         for _ in range(5):
 860             res = self.sendUDPQuery(query)
 861             self.assertRcodeEqual(res, dns.rcode.NOERROR)
 862             self.assertRRsetInAnswer(res, first.answer[0])
 863
 864     def testTimeout(self):
 865         """
 866         Test if LUA scripts are aborted if script execution takes too long
 867         """
 868         query = dns.message.make_query('timeout.example.org', 'A')
 869
 870         first = self.sendUDPQuery(query)
 871         self.assertRcodeEqual(first, dns.rcode.SERVFAIL)
 872
 873
 874     def testA(self):
 875         """
 876         Test A query against `any`
 877         """
 878         name = 'any.example.org.'
 879
 880         query = dns.message.make_query(name, 'A')
 881
 882         response = dns.message.make_response(query)
 883
 884         response.answer.append(dns.rrset.from_text(name, 0, dns.rdataclass.IN, dns.rdatatype.A, '192.0.2.1'))
 885
 886         res = self.sendUDPQuery(query)
 887         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 888         self.assertEqual(res.answer, response.answer)
 889
 890     def testANY(self):
 891         """
 892         Test ANY query against `any`
 893         """
 894
 895         name = 'any.example.org.'
 896
 897         query = dns.message.make_query(name, 'ANY')
 898
 899         response = dns.message.make_response(query)
 900
 901         response.answer.append(dns.rrset.from_text(name, 0, dns.rdataclass.IN, dns.rdatatype.A, '192.0.2.1'))
 902         response.answer.append(dns.rrset.from_text(name, 0, dns.rdataclass.IN, 'TXT', '"hello there"'))
 903
 904         res = self.sendUDPQuery(query)
 905         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 906         self.assertEqual(self.sortRRsets(res.answer), self.sortRRsets(response.answer))
 907
 908     def testCAFromRaw(self):
 909         """
 910         Test newCAFromRaw() function
 911         """
 912         name = 'newcafromraw.example.org.'
 913
 914         query = dns.message.make_query(name, 'A')
 915
 916         response = dns.message.make_response(query)
 917
 918         response.answer.append(dns.rrset.from_text(name, 0, dns.rdataclass.IN, dns.rdatatype.A, '65.66.67.68'))
 919
 920         res = self.sendUDPQuery(query)
 921         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 922         self.assertEqual(res.answer, response.answer)
 923
 924         query = dns.message.make_query(name, 'AAAA')
 925
 926         response = dns.message.make_response(query)
 927
 928         response.answer.append(dns.rrset.from_text(name, 0, dns.rdataclass.IN, dns.rdatatype.AAAA, '4142:4344:3032:3033:3430:3530:3630:3730'))
 929
 930         res = self.sendUDPQuery(query)
 931         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 932         self.assertEqual(res.answer, response.answer)
 933
 934     def testResolve(self):
 935         """
 936         Test resolve() function
 937         """
 938         name = 'resolve.example.org.'
 939
 940         query = dns.message.make_query(name, 'A')
 941
 942         response = dns.message.make_response(query)
 943
 944         response.answer.append(dns.rrset.from_text(name, 0, dns.rdataclass.IN, dns.rdatatype.A, '127.0.0.1'))
 945
 946         res = self.sendUDPQuery(query)
 947         self.assertRcodeEqual(res, dns.rcode.NOERROR)
 948         self.assertEqual(res.answer, response.answer)
 949
 950     def testCreateForwardAndReverse(self):
 951         expected = {
 952             ".createforward.example.org." : (dns.rdatatype.A, {
 953                 "1.2.3.4": "1.2.3.4",
 954                 "1.2.3.4.static": "1.2.3.4",
 955                 "1.2.3.4.5.6": "1.2.3.4",
 956                 "invalid.1.2.3.4": "0.0.0.0",
 957                 "invalid": "0.0.0.0",
 958                 "1-2-3-4": "1.2.3.4",
 959                 "1-2-3-4.foo": "1.2.3.4",
 960                 "1-2-3-4.foo.bar": "0.0.0.0",
 961                 "1-2-3-4.foo.bar.baz": "0.0.0.0",
 962                 "1-2-3-4.foo.bar.baz.quux": "0.0.0.0",
 963                 "ip-1-2-3-4": "1.2.3.4",
 964                 "ip-is-here-for-you-1-2-3-4": "1.2.3.4",
 965                 "40414243": "64.65.66.67",
 966                 "p40414243": "64.65.66.67",
 967                 "ip40414243": "64.65.66.67",
 968                 "ipp40414243": "64.65.66.67",
 969                 "ip4041424": "0.0.0.0",
 970                 "host64-22-33-44": "64.22.33.44",
 971                 "2.2.2.2": "0.0.0.0"   # filtered
 972             }),
 973             ".createreverse.example.org." : (dns.rdatatype.PTR, {
 974                 "4.3.2.1": "1-2-3-4.example.com.",
 975                 "10.10.10.10": "quad10.example.com."   # exception
 976             }),
 977             ".createforward6.example.org." : (dns.rdatatype.AAAA, {
 978                 "2001--db8" : "2001::db8",
 979                 "20010002000300040005000600070db8" : "2001:2:3:4:5:6:7:db8",
 980                 "blabla20010002000300040005000600070db8" : "2001:2:3:4:5:6:7:db8",
 981                 "4000-db8--1" : "fe80::1"   # filtered, with fallback address override
 982             }),
 983             ".createreverse6.example.org." : (dns.rdatatype.PTR, {
 984                 "8.b.d.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.1.0.0.2" : "2001--db8.example.com.",
 985                 "1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.8.b.d.0.1.0.0.2" : "example.example.com."   # exception
 986             })
 987         }
 988
 989         for suffix, v in expected.items():
 990             qtype, pairs = v
 991             for prefix, target in pairs.items():
 992                 name = prefix + suffix
 993
 994                 query = dns.message.make_query(name, qtype)
 995                 response = dns.message.make_response(query)
 996                 response.answer.append(dns.rrset.from_text(
 997                     name, 0, dns.rdataclass.IN, qtype, target))
 998
 999                 res = self.sendUDPQuery(query)
1000                 print(res)
1001                 self.assertRcodeEqual(res, dns.rcode.NOERROR)
1002                 self.assertEqual(res.answer, response.answer)
1003
1004     def _getCounter(self, tcp=False):
1005         """
1006         Helper function for shared/non-shared testing
1007         """
1008         name = 'counter.example.org.'
1009
1010         query = dns.message.make_query(name, 'TXT')
1011         responses = []
1012
1013         sender = self.sendTCPQuery if tcp else self.sendUDPQuery
1014
1015         for i in range(50):
1016             res = sender(query)
1017             responses.append(res.answer[0][0])
1018
1019         return(responses)
1020
1021     def testCounter(self):
1022         """
1023         Test non-shared behaviour
1024         """
1025
1026         resUDP = set(self._getCounter(tcp=False))
1027         resTCP = set(self._getCounter(tcp=True))
1028
1029         self.assertEqual(len(resUDP), 1)
1030         self.assertEqual(len(resTCP), 1)
1031
1032     def testDblookup(self):
1033         """
1034         Test dblookup() function
1035         """
1036
1037         name = 'dblookup.example.org.'
1038
1039         query = dns.message.make_query(name, 'A')
1040
1041         response = dns.message.make_response(query)
1042
1043         response.answer.append(dns.rrset.from_text(name, 0, dns.rdataclass.IN, 'A', '192.0.2.5'))
1044
1045         res = self.sendUDPQuery(query)
1046         self.assertRcodeEqual(res, dns.rcode.NOERROR)
1047         self.assertEqual(self.sortRRsets(res.answer), self.sortRRsets(response.answer))
1048
1049
1050 class TestLuaRecordsShared(TestLuaRecords):
1051     _config_template = """
1052 geoip-database-files=../modules/geoipbackend/regression-tests/GeoLiteCity.mmdb
1053 edns-subnet-processing=yes
1054 launch=bind geoip
1055 any-to-tcp=no
1056 enable-lua-records=shared
1057 lua-health-checks-interval=1
1058 """
1059
1060     def testCounter(self):
1061         """
1062         Test shared behaviour
1063         """
1064
1065         resUDP = set(self._getCounter(tcp=False))
1066         resTCP = set(self._getCounter(tcp=True))
1067
1068         self.assertEqual(len(resUDP), 50)
1069         self.assertEqual(len(resTCP), 50)
1070
1071 if __name__ == '__main__':
1072     unittest.main()
1073     exit(0)