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
|
import unittest
import time
import lib
lib.setup_env()
from common import resolver
from mock import Mock, expectParams
from gajim_mocks import *
from xmpp_mocks import IdleQueueThread
GMAIL_SRV_NAME = '_xmpp-client._tcp.gmail.com'
NONSENSE_NAME = 'sfsdfsdfsdf.sdfs.fsd'
JABBERCZ_TXT_NAME = '_xmppconnect.jabber.cz'
JABBERCZ_SRV_NAME = '_xmpp-client._tcp.jabber.cz'
TEST_LIST = [(GMAIL_SRV_NAME, 'srv', True),
(NONSENSE_NAME, 'srv', False),
(JABBERCZ_SRV_NAME, 'srv', True)]
class TestResolver(unittest.TestCase):
'''
Test for LibAsyncNSResolver and NSLookupResolver. Requires working
network connection.
'''
def setUp(self):
self.idlequeue_thread = IdleQueueThread()
self.idlequeue_thread.start()
self.iq = self.idlequeue_thread.iq
self._reset()
self.resolver = None
def tearDown(self):
self.idlequeue_thread.stop_thread()
self.idlequeue_thread.join()
def _reset(self):
self.flag = False
self.expect_results = False
self.nslookup = False
self.resolver = None
def testLibAsyncNSResolver(self):
self._reset()
if not resolver.USE_LIBASYNCNS:
print 'testLibAsyncResolver: libasyncns-python not installed'
return
self.resolver = resolver.LibAsyncNSResolver()
for name, type, expect_results in TEST_LIST:
self.expect_results = expect_results
self._runLANSR(name, type)
self.flag = False
def _runLANSR(self, name, type):
self.resolver.resolve(
host = name,
type = type,
on_ready = self._myonready)
while not self.flag:
time.sleep(1)
self.resolver.process()
def _myonready(self, name, result_set):
if __name__ == '__main__':
from pprint import pprint
pprint('on_ready called ...')
pprint('hostname: %s' % name)
pprint('result set: %s' % result_set)
pprint('res.resolved_hosts: %s' % self.resolver.resolved_hosts)
pprint('')
if self.expect_results:
self.assert_(len(result_set) > 0)
else:
self.assert_(result_set == [])
self.flag = True
if self.nslookup:
self._testNSLR()
def testNSLookupResolver(self):
self._reset()
self.nslookup = True
self.resolver = resolver.NSLookupResolver(self.iq)
self.test_list = TEST_LIST
self._testNSLR()
def _testNSLR(self):
if self.test_list == []:
return
name, type, self.expect_results = self.test_list.pop()
self.resolver.resolve(
host = name,
type = type,
on_ready = self._myonready)
if __name__ == '__main__':
unittest.main()
|
