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_channel_timeout(self):
self.result, self.errorno = None, None
def cb(result, errorno):
self.result, self.errorno = result, errorno
self.channel = pycares.Channel(timeout=0.5, tries=1)
self.channel.gethostbyname('google.com', socket.AF_INET, cb)
timeout = self.channel.timeout()
self.assertTrue(timeout > 0.0)
self.channel.cancel()
self.wait()
self.assertEqual(self.result, None)
self.assertEqual(self.errorno, pycares.errno.ARES_ECANCELLED)
def test_idna_encoding(self):
host = 'españa.icom.museum'
self.result, self.errorno = None, None
def cb(result, errorno):
self.result, self.errorno = result, errorno
# try encoding it as utf-8
self.channel.gethostbyname(host.encode(), socket.AF_INET, cb)
self.wait()
self.assertEqual(self.errorno, pycares.errno.ARES_ENOTFOUND)
self.assertEqual(self.result, None)
# use it as is (it's IDNA encoded internally)
self.channel.gethostbyname(host, socket.AF_INET, cb)
self.wait()
self.assertNoError(self.errorno)
self.assertEqual(type(self.result), pycares.ares_host_result)
def test_errorcode_dict(self):
for err in ('ARES_SUCCESS', 'ARES_ENODATA', 'ARES_ECANCELLED'):
val = getattr(pycares.errno, err)
self.assertEqual(pycares.errno.errorcode[val], err)
def resolve(self, host, port, family=0):
if is_valid_ip(host):
addresses = [host]
else:
# gethostbyname doesn't take callback as a kwarg
self.channel.gethostbyname(host, family, (yield gen.Callback(1)))
callback_args = yield gen.Wait(1)
assert isinstance(callback_args, gen.Arguments)
assert not callback_args.kwargs
result, error = callback_args.args
if error:
raise IOError('C-Ares returned error %s: %s while resolving %s' %
(error, pycares.errno.strerror(error), host))
addresses = result.addresses
addrinfo = []
for address in addresses:
if '.' in address:
address_family = socket.AF_INET
elif ':' in address:
address_family = socket.AF_INET6
else:
address_family = socket.AF_UNSPEC
if family != socket.AF_UNSPEC and family != address_family:
raise IOError('Requested socket family %d but got %d' %
(family, address_family))
addrinfo.append((address_family, (address, port)))
raise gen.Return(addrinfo)
def _callback(fut: asyncio.Future, result: Any, errorno: int) -> None:
if fut.cancelled():
return
if errorno is not None:
fut.set_exception(error.DNSError(errorno, pycares.errno.strerror(errorno)))
else:
fut.set_result(result)
def resolve(self, host, port, family=0):
if is_valid_ip(host):
addresses = [host]
else:
# gethostbyname doesn't take callback as a kwarg
fut = Future()
self.channel.gethostbyname(host, family,
lambda result, error: fut.set_result((result, error)))
result, error = yield fut
if error:
raise IOError('C-Ares returned error %s: %s while resolving %s' %
(error, pycares.errno.strerror(error), host))
addresses = result.addresses
addrinfo = []
for address in addresses:
if '.' in address:
address_family = socket.AF_INET
elif ':' in address:
address_family = socket.AF_INET6
else:
address_family = socket.AF_UNSPEC
if family != socket.AF_UNSPEC and family != address_family:
raise IOError('Requested socket family %d but got %d' %
(family, address_family))
addrinfo.append((address_family, (address, port)))
raise gen.Return(addrinfo)
def _ares_cb2(self, cb, result, error):
if error is not None:
error_data = self._ares_errno_map2.get(error)
if not error_data:
exc = socket.gaierror(error, pycares.errno.strerror(error))
else:
klass, args = error_data
exc = klass(*args)
cb(Result(None, exc))
else:
cb(Result(result, None))
def __init__(self, errno):
str_error = pycares.errno.strerror(errno)
if isinstance(str_error, bytes):
str_error = str_error.decode('utf-8')
msg = '{0} [{1}]'.format(str_error, pycares.errno.errorcode[errno])
super(DNSError, self).__init__(msg)
self.errno = errno