1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
import socket
import sys
import unittest
import dns.name
import dns.rdataclass
import dns.rdatatype
import dns.resolver
# Some tests require the internet to be available to run, so let's
# skip those if it's not there.
_network_available = True
try:
socket.gethostbyname('dnspython.org')
except socket.gaierror:
_network_available = False
@unittest.skipIf(not _network_available, "Internet not reachable")
class OverrideSystemResolverTestCase(unittest.TestCase):
def setUp(self):
self.res = dns.resolver.Resolver()
self.res.nameservers = ['8.8.8.8']
self.res.cache = dns.resolver.LRUCache()
dns.resolver.override_system_resolver(self.res)
def tearDown(self):
dns.resolver.restore_system_resolver()
self.res = None
def test_override(self):
self.assertTrue(socket.getaddrinfo is
dns.resolver._getaddrinfo)
socket.gethostbyname('www.dnspython.org')
answer = self.res.cache.get((dns.name.from_text('www.dnspython.org.'),
dns.rdatatype.A, dns.rdataclass.IN))
self.assertTrue(answer is not None)
self.res.cache.flush()
socket.gethostbyname_ex('www.dnspython.org')
answer = self.res.cache.get((dns.name.from_text('www.dnspython.org.'),
dns.rdatatype.A, dns.rdataclass.IN))
self.assertTrue(answer is not None)
self.res.cache.flush()
socket.getfqdn('8.8.8.8')
answer = self.res.cache.get(
(dns.name.from_text('8.8.8.8.in-addr.arpa.'),
dns.rdatatype.PTR, dns.rdataclass.IN))
self.assertTrue(answer is not None)
self.res.cache.flush()
socket.gethostbyaddr('8.8.8.8')
answer = self.res.cache.get(
(dns.name.from_text('8.8.8.8.in-addr.arpa.'),
dns.rdatatype.PTR, dns.rdataclass.IN))
self.assertTrue(answer is not None)
# restoring twice is harmless, so we restore now instead of
# waiting for tearDown so we can assert that it worked
dns.resolver.restore_system_resolver()
self.assertTrue(socket.getaddrinfo is
dns.resolver._original_getaddrinfo)
def equivalent_info(self, a, b):
if len(a) != len(b):
return False
for x in a:
if x not in b:
# Windows does not set the protocol to non-zero, so try
# looking for a zero protocol.
y = (x[0], x[1], 0, x[3], x[4])
if y not in b:
print('NOT EQUIVALENT')
print(a)
print(b)
return False
return True
def equivalent(self, *args, **kwargs):
a = socket.getaddrinfo(*args, **kwargs)
b = dns.resolver._original_getaddrinfo(*args, **kwargs)
return self.equivalent_info(a, b)
@unittest.skipIf(sys.platform == 'win32',
'avoid windows original getaddrinfo issues')
def test_basic_getaddrinfo(self):
self.assertTrue(self.equivalent('dns.google', 53, socket.AF_INET,
socket.SOCK_DGRAM))
self.assertTrue(self.equivalent('dns.google', 53, socket.AF_INET6,
socket.SOCK_DGRAM))
self.assertTrue(self.equivalent('dns.google', None, socket.AF_UNSPEC,
socket.SOCK_DGRAM))
self.assertTrue(self.equivalent('8.8.8.8', 53, socket.AF_INET,
socket.SOCK_DGRAM))
self.assertTrue(self.equivalent('2001:4860:4860::8888', 53,
socket.AF_INET6, socket.SOCK_DGRAM))
self.assertTrue(self.equivalent('8.8.8.8', 53, socket.AF_INET,
socket.SOCK_DGRAM,
flags=socket.AI_NUMERICHOST))
self.assertTrue(self.equivalent('2001:4860:4860::8888', 53,
socket.AF_INET6, socket.SOCK_DGRAM,
flags=socket.AI_NUMERICHOST))
def test_getaddrinfo_nxdomain(self):
try:
socket.getaddrinfo('nxdomain.dnspython.org.', 53)
self.assertTrue(False) # should not happen!
except socket.gaierror as e:
self.assertEqual(e.errno, socket.EAI_NONAME)
def test_getaddrinfo_service(self):
a = socket.getaddrinfo('dns.google', 'domain')
b = socket.getaddrinfo('dns.google', 53)
self.assertTrue(self.equivalent_info(a, b))
try:
socket.getaddrinfo('dns.google', 'domain',
flags=socket.AI_NUMERICSERV)
self.assertTrue(False) # should not happen!
except socket.gaierror as e:
self.assertEqual(e.errno, socket.EAI_NONAME)
def test_getfqdn(self):
b = socket.getfqdn('www.dnspython.org')
# we do this now because python's original getfqdn calls
# gethostbyaddr() and we don't want it to call us!
dns.resolver.restore_system_resolver()
a = dns.resolver._original_getfqdn('www.dnspython.org')
self.assertEqual(dns.name.from_text(a), dns.name.from_text(b))
def test_gethostbyaddr(self):
a = dns.resolver._original_gethostbyaddr('8.8.8.8')
b = socket.gethostbyaddr('8.8.8.8')
# We only test elements 0 and 2 as we don't set aliases currently!
self.assertEqual(a[0], b[0])
self.assertEqual(a[2], b[2])
a = dns.resolver._original_gethostbyaddr('2001:4860:4860::8888')
b = socket.gethostbyaddr('2001:4860:4860::8888')
self.assertEqual(a[0], b[0])
self.assertEqual(a[2], b[2])
|
