Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_query_timeout(self):
self.result, self.errorno = None, None
def cb(result, errorno):
self.result, self.errorno = result, errorno
self.channel.servers = ['1.2.3.4']
self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
self.wait()
self.assertEqual(self.result, None)
self.assertEqual(self.errorno, pycares.errno.ARES_ETIMEOUT)
def test_search(self):
self.result, self.errorno = None, None
def cb(result, errorno):
self.result, self.errorno = result, errorno
self.channel = pycares.Channel(timeout=5.0, tries=1, domains=['google.com'])
self.channel.search('cloud', pycares.QUERY_TYPE_A, cb)
self.wait()
self.assertNoError(self.errorno)
for r in self.result:
self.assertEqual(type(r), pycares.ares_query_a_result)
self.assertNotEqual(r.host, None)
def test_custom_resolvconf(self):
self.result, self.errorno = None, None
def cb(result, errorno):
self.result, self.errorno = result, errorno
self.channel = pycares.Channel(tries=1, timeout=2.0, resolvconf_path=os.path.join(FIXTURES_PATH, 'badresolv.conf'))
self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
self.wait()
self.assertEqual(self.result, None)
self.assertEqual(self.errorno, pycares.errno.ARES_ETIMEOUT)
def test_query_a_rotate(self):
self.result, self.errorno = None, None
self.errorno_count, self.count = 0, 0
def cb(result, errorno):
self.result, self.errorno = result, errorno
if errorno:
self.errorno_count += 1
self.count += 1
self.channel = pycares.Channel(timeout=1.0, tries=1, rotate=True)
self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
self.wait()
self.assertEqual(self.count, 3)
self.assertEqual(self.errorno_count, 0)
thread = threading.Thread(target=_query_pan_dns, args=(parent_domain,))
thread.start()
thread.join()
record = None
for query_type in ("A", "CNAME"):
try:
with async_timeout.timeout(_Timeout):
records = await _resolver.query(domain, query_type)
except:
pass
else:
if query_type == "CNAME":
record = Record(domain, QUERY_TYPE_CNAME, records.ttl if records.ttl else -1, [records.cname])
elif query_type == "A" and records:
record = Record(domain, QUERY_TYPE_A, records[0].ttl, [record_.host for record_ in records])
if record:
break
if record:
return None if _is_pan_dns(record) else record
def __init__(self, name, channel, res):
def cb(results, err):
if results is None:
return
for result in results:
res.add((result.host, self.name))
self.name = name
channel.query(name, pycares.QUERY_TYPE_A, cb)
Optional,
Set
)
from . import error
__version__ = '2.0.0'
__all__ = ('DNSResolver', 'error')
READ = 1
WRITE = 2
query_type_map = {'A' : pycares.QUERY_TYPE_A,
'AAAA' : pycares.QUERY_TYPE_AAAA,
'ANY' : pycares.QUERY_TYPE_ANY,
'CNAME' : pycares.QUERY_TYPE_CNAME,
'MX' : pycares.QUERY_TYPE_MX,
'NAPTR' : pycares.QUERY_TYPE_NAPTR,
'NS' : pycares.QUERY_TYPE_NS,
'PTR' : pycares.QUERY_TYPE_PTR,
'SOA' : pycares.QUERY_TYPE_SOA,
'SRV' : pycares.QUERY_TYPE_SRV,
'TXT' : pycares.QUERY_TYPE_TXT
}
class DNSResolver:
def __init__(self, nameservers: Optional[List[str]] = None, loop: Optional[asyncio.AbstractEventLoop] = None,
**kwargs: Any) -> None:
def gethostbyname(self, name, cb):
self._channel.gethostbyname(name, socket.AF_INET, cb)
if __name__ == '__main__':
def query_cb(result, error):
print(result)
print(error)
def gethostbyname_cb(result, error):
print(result)
print(error)
resolver = DNSResolver()
resolver.query('google.com', pycares.QUERY_TYPE_A, query_cb)
resolver.query('google.com', pycares.QUERY_TYPE_AAAA, query_cb)
resolver.query('facebook.com', pycares.QUERY_TYPE_A, query_cb)
resolver.query('sip2sip.info', pycares.QUERY_TYPE_SOA, query_cb)
resolver.gethostbyname('apple.com', gethostbyname_cb)
resolver.wait_channel()
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:
channel.process_fd(fd, pycares.ARES_SOCKET_BAD)
for fd in wlist:
channel.process_fd(pycares.ARES_SOCKET_BAD, fd)
if __name__ == '__main__':
def cb(result, error):
print(result)
print(error)
channel = pycares.Channel()
channel.gethostbyname('google.com', socket.AF_INET, cb)
channel.query('google.com', pycares.QUERY_TYPE_A, cb)
channel.query('sip2sip.info', pycares.QUERY_TYPE_SOA, cb)
wait_channel(channel)
print('Done!')