/
test_dns.py
155 lines (125 loc) · 5.04 KB
/
test_dns.py
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
from common import unittest2
import pycares
import select
import socket
class DNSTest(unittest2.TestCase):
def setUp(self):
self.channel = pycares.Channel(timeout=1.0, tries=1)
def tearDown(self):
self.channel = None
def wait(self):
while True:
read_fds, write_fds = self.channel.getsock()
if not read_fds and not write_fds:
break
timeout = self.channel.timeout()
if timeout == 0.0:
self.channel.process_fd(pycares.ARES_SOCKET_BAD, pycares.ARES_SOCKET_BAD)
continue
rlist, wlist, xlist = select.select(read_fds, write_fds, [], timeout)
for fd in rlist:
self.channel.process_fd(fd, pycares.ARES_SOCKET_BAD)
for fd in wlist:
self.channel.process_fd(pycares.ARES_SOCKET_BAD, fd)
def test_gethostbyaddr(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.gethostbyaddr('127.0.0.1', cb)
self.wait()
def test_gethostbyaddr6(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.gethostbyaddr('::1', cb)
self.wait()
def test_gethostbyname(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.gethostbyname('localhost', socket.AF_INET, cb)
self.wait()
def test_getnameinfo(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.assertEqual(result, ('localhost', 'http'))
self.channel.getnameinfo(('127.0.0.1', 80), pycares.ARES_NI_LOOKUPHOST|pycares.ARES_NI_LOOKUPSERVICE, cb)
self.wait()
def test_query_a(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
self.wait()
def test_query_a_bad(self):
def cb(result, errorno):
self.assertEqual(result, None)
self.assertEqual(errorno, pycares.errno.ARES_ENOTFOUND)
self.channel.query('hgf8g2od29hdohid.com', pycares.QUERY_TYPE_A, cb)
self.wait()
def test_query_aaaa(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.query('ipv6.google.com', pycares.QUERY_TYPE_AAAA, cb)
self.wait()
def test_query_cname(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.query('www.google.com', pycares.QUERY_TYPE_CNAME, cb)
self.wait()
def test_query_mx(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.query('google.com', pycares.QUERY_TYPE_MX, cb)
self.wait()
def test_query_ns(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.query('google.com', pycares.QUERY_TYPE_NS, cb)
self.wait()
def test_query_txt(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.query('google.com', pycares.QUERY_TYPE_TXT, cb)
self.wait()
def test_query_soa(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.query('google.com', pycares.QUERY_TYPE_SOA, cb)
self.wait()
def test_query_srv(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.query('_xmpp-server._tcp.google.com', pycares.QUERY_TYPE_SRV, cb)
self.wait()
def test_query_naptr(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
self.channel.query('sip2sip.info', pycares.QUERY_TYPE_NAPTR, cb)
self.wait()
def test_query_ptr(self):
def cb(result, errorno):
self.assertEqual(errorno, None)
ip = '173.194.69.102'
self.channel.query(pycares.reverse_address(ip), pycares.QUERY_TYPE_PTR, cb)
self.wait()
def test_query_cancelled(self):
def cb(result, errorno):
self.assertEqual(errorno, pycares.errno.ARES_ECANCELLED)
self.channel.query('google.com', pycares.QUERY_TYPE_NS, cb)
self.channel.cancel()
self.wait()
def test_channel_destroyed(self):
self.channel.destroy()
self.assertRaises(pycares.AresError, self.channel.gethostbyname, 'google.com', socket.AF_INET, lambda *x: None)
def test_query_bad_type(self):
self.assertRaises(pycares.AresError, self.channel.query, 'google.com', 667, lambda *x: None)
self.wait()
def test_query_timeout(self):
def cb(result, errorno):
self.assertEqual(errorno, pycares.errno.ARES_ETIMEOUT)
self.channel.servers = ['1.2.3.4']
self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
self.wait()
def test_reverse_address(self):
s = '1.2.3.4'
expected = '4.3.2.1.in-addr.arpa'
self.assertEqual(pycares.reverse_address(s), expected)
if __name__ == '__main__':
unittest2.main(verbosity=2)