import dns
import requests
import socket
+import time
+import extendederrors
+
from recursortests import RecursorTest
class RootNXTrustRecursorTest(RecursorTest):
url = 'http://127.0.0.1:' + str(self._wsPort) + '/api/v1/servers/localhost/statistics'
r = requests.get(url, headers=headers, timeout=self._wsTimeout)
self.assertTrue(r)
- self.assertEquals(r.status_code, 200)
+ self.assertEqual(r.status_code, 200)
self.assertTrue(r.json())
content = r.json()
for entry in content:
return 0
+ # Recursor can still be busy resolving root hints, so wait a bit until
+ # getOutgoingQueriesCount() stabilizes.
+ # Code below is inherently racey, but better than a fixed sleep
+ def waitForOutgoingToStabilize(self):
+ for count in range(20):
+ outgoing1 = self.getOutgoingQueriesCount();
+ time.sleep(0.1);
+ outgoing2 = self.getOutgoingQueriesCount();
+ if outgoing1 == outgoing2:
+ break
+
class testRootNXTrustDisabled(RootNXTrustRecursorTest):
_confdir = 'RootNXTrustDisabled'
_wsPort = 8042
webserver-address=127.0.0.1
webserver-password=%s
api-key=%s
+devonly-regression-test-mode
+extended-resolution-errors
""" % (_wsPort, _wsPassword, _apiKey)
def testRootNXTrust(self):
after receiving a NXD from "." for nx-example. as an answer for www.nx-example.
"""
- # first query nx.example.
+ self.waitForTCPSocket("127.0.0.1", self._wsPort)
+ self.waitForOutgoingToStabilize()
+ # First query nx.example.
before = self.getOutgoingQueriesCount()
query = dns.message.make_query('www.nx-example.', 'A')
res = self.sendUDPQuery(query)
# then query nx2.example.
before = after
- query = dns.message.make_query('www2.nx-example.', 'A')
+ query = dns.message.make_query('www2.nx-example.', 'A', use_edns=True)
res = self.sendUDPQuery(query)
self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
after = self.getOutgoingQueriesCount()
self.assertEqual(after, before + 1)
+ self.assertEqual(res.edns, 0)
+ self.assertEqual(len(res.options), 0)
class testRootNXTrustEnabled(RootNXTrustRecursorTest):
_confdir = 'RootNXTrustEnabled'
webserver-address=127.0.0.1
webserver-password=%s
api-key=%s
+devonly-regression-test-mode
+extended-resolution-errors
""" % (_wsPort, _wsPassword, _apiKey)
def testRootNXTrust(self):
after receiving a NXD from "." for nx-example. as an answer for www.nx-example.
"""
+ self.waitForTCPSocket("127.0.0.1", self._wsPort)
+ self.waitForOutgoingToStabilize()
# first query nx.example.
before = self.getOutgoingQueriesCount()
query = dns.message.make_query('www.nx-example.', 'A')
# then query nx2.example.
before = after
- query = dns.message.make_query('www2.nx-example.', 'A')
+ query = dns.message.make_query('www2.nx-example.', 'A', use_edns=True)
res = self.sendUDPQuery(query)
self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
after = self.getOutgoingQueriesCount()
self.assertEqual(after, before)
+ self.assertEqual(res.edns, 0)
+ self.assertEqual(len(res.options), 1)
+ self.assertEqual(res.options[0].otype, 15)
+ self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized by root-nx-trust'))