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()