run: |
python -m pip install --upgrade pip
python -m pip install poetry
- poetry install -E dnssec -E doh -E idna -E trio -E curio -E doq
+ poetry install -E dnssec -E doh -E idna -E trio -E doq
- name: Typecheck
run: |
poetry run python -m mypy --install-types --non-interactive --disallow-incomplete-defs dns
+++ /dev/null
-# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
-
-"""curio async I/O library query support"""
-
-import socket
-import curio
-import curio.socket # type: ignore
-
-import dns._asyncbackend
-import dns.exception
-import dns.inet
-
-
-def _maybe_timeout(timeout):
- if timeout:
- return curio.ignore_after(timeout)
- else:
- return dns._asyncbackend.NullContext()
-
-
-# for brevity
-_lltuple = dns.inet.low_level_address_tuple
-
-# pylint: disable=redefined-outer-name
-
-
-class DatagramSocket(dns._asyncbackend.DatagramSocket):
- def __init__(self, socket):
- super().__init__(socket.family)
- self.socket = socket
-
- async def sendto(self, what, destination, timeout):
- async with _maybe_timeout(timeout):
- return await self.socket.sendto(what, destination)
- raise dns.exception.Timeout(
- timeout=timeout
- ) # pragma: no cover lgtm[py/unreachable-statement]
-
- async def recvfrom(self, size, timeout):
- async with _maybe_timeout(timeout):
- return await self.socket.recvfrom(size)
- raise dns.exception.Timeout(timeout=timeout) # lgtm[py/unreachable-statement]
-
- async def close(self):
- await self.socket.close()
-
- async def getpeername(self):
- return self.socket.getpeername()
-
- async def getsockname(self):
- return self.socket.getsockname()
-
-
-class StreamSocket(dns._asyncbackend.StreamSocket):
- def __init__(self, socket):
- self.socket = socket
- self.family = socket.family
-
- async def sendall(self, what, timeout):
- async with _maybe_timeout(timeout):
- return await self.socket.sendall(what)
- raise dns.exception.Timeout(timeout=timeout) # lgtm[py/unreachable-statement]
-
- async def recv(self, size, timeout):
- async with _maybe_timeout(timeout):
- return await self.socket.recv(size)
- raise dns.exception.Timeout(timeout=timeout) # lgtm[py/unreachable-statement]
-
- async def close(self):
- await self.socket.close()
-
- async def getpeername(self):
- return self.socket.getpeername()
-
- async def getsockname(self):
- return self.socket.getsockname()
-
-
-class Backend(dns._asyncbackend.Backend):
- def name(self):
- return "curio"
-
- async def make_socket(
- self,
- af,
- socktype,
- proto=0,
- source=None,
- destination=None,
- timeout=None,
- ssl_context=None,
- server_hostname=None,
- ):
- if socktype == socket.SOCK_DGRAM:
- s = curio.socket.socket(af, socktype, proto)
- try:
- if source:
- s.bind(_lltuple(source, af))
- except Exception: # pragma: no cover
- await s.close()
- raise
- return DatagramSocket(s)
- elif socktype == socket.SOCK_STREAM:
- if source:
- source_addr = _lltuple(source, af)
- else:
- source_addr = None
- async with _maybe_timeout(timeout):
- s = await curio.open_connection(
- destination[0],
- destination[1],
- ssl=ssl_context,
- source_addr=source_addr,
- server_hostname=server_hostname,
- )
- return StreamSocket(s)
- raise NotImplementedError(
- "unsupported socket " + f"type {socktype}"
- ) # pragma: no cover
-
- async def sleep(self, interval):
- await curio.sleep(interval)
def get_backend(name: str) -> Backend:
"""Get the specified asynchronous backend.
- *name*, a ``str``, the name of the backend. Currently the "trio",
- "curio", and "asyncio" backends are available.
+ *name*, a ``str``, the name of the backend. Currently the "trio"
+ and "asyncio" backends are available.
Raises NotImplementError if an unknown backend name is specified.
"""
import dns._trio_backend
backend = dns._trio_backend.Backend()
- elif name == "curio":
- import dns._curio_backend
-
- backend = dns._curio_backend.Backend()
elif name == "asyncio":
import dns._asyncio_backend
class Zone(dns.zone.Zone): # lgtm[py/missing-equals]
-
__slots__ = [
"_versions",
"_versions_lock",
#
# This is not a problem with Threading module threads as
# they cannot be canceled, but could be an issue with trio
- # or curio tasks when we do the async version of writer().
+ # tasks when we do the async version of writer().
# I.e. we'd need to do something like:
#
# try:
[mypy-requests_toolbelt.*]
ignore_missing_imports = True
-[mypy-curio]
-ignore_missing_imports = True
-
[mypy-trio]
ignore_missing_imports = True
idna = {version=">=2.1,<4.0", optional=true}
cryptography = {version=">=2.6,<40.0", optional=true}
trio = {version=">=0.14,<0.23", optional=true}
-curio = {version="^1.2", optional=true}
sniffio = {version="^1.1", optional=true}
wmi = {version="^1.5.1", optional=true}
aioquic = {version=">=0.9.20", optional=true}
idna = ['idna']
dnssec = ['cryptography']
trio = ['trio']
-curio = ['curio', 'sniffio']
wmi = ['wmi']
doq = ['aioquic']
IDNA = idna>=2.1
DNSSEC = cryptography>=2.6
trio = trio>=0.14.0
-curio = curio>=1.2; sniffio>=1.1
wmi = wmi>=1.5.1
DOQ = aioquic>=0.9.20
self.async_run(run4)
async def run5():
- await dns.asyncresolver.resolve_name(dns.reversename.from_address("8.8.8.8"))
+ await dns.asyncresolver.resolve_name(
+ dns.reversename.from_address("8.8.8.8")
+ )
with self.assertRaises(dns.resolver.NoAnswer):
self.async_run(run5)
@unittest.skipIf(not dns.query._have_httpx, "httpx not available")
def testDOHGetRequest(self):
- if self.backend.name() == "curio":
- self.skipTest("anyio dropped curio support")
-
async def run():
nameserver_url = random.choice(KNOWN_ANYCAST_DOH_RESOLVER_URLS)
q = dns.message.make_query("example.com.", dns.rdatatype.A)
@unittest.skipIf(not dns.query._have_httpx, "httpx not available")
def testDOHGetRequestHttp1(self):
- if self.backend.name() == "curio":
- self.skipTest("anyio dropped curio support")
-
async def run():
saved_have_http2 = dns.query._have_http2
try:
@unittest.skipIf(not dns.query._have_httpx, "httpx not available")
def testDOHPostRequest(self):
- if self.backend.name() == "curio":
- self.skipTest("anyio dropped curio support")
-
async def run():
nameserver_url = random.choice(KNOWN_ANYCAST_DOH_RESOLVER_URLS)
q = dns.message.make_query("example.com.", dns.rdatatype.A)
@unittest.skipIf(not dns.query._have_httpx, "httpx not available")
def testResolverDOH(self):
- if self.backend.name() == "curio":
- self.skipTest("anyio dropped curio support")
-
async def run():
res = dns.asyncresolver.Resolver(configure=False)
res.nameservers = ["https://dns.google/dns-query"]
def testUseAfterTimeout(self):
if self.connect_udp:
self.skipTest("test needs connectionless sockets")
+
# Test #843 fix.
async def run():
qname = dns.name.from_text("dns.google")
except ImportError:
pass
-
-try:
- import curio
- import sniffio
-
- @unittest.skipIf(sys.platform == "win32", "curio does not work in windows CI")
- class CurioAsyncDetectionTests(AsyncDetectionTests):
- sniff_result = "curio"
-
- def async_run(self, afunc):
- with curio.Kernel() as kernel:
- return kernel.run(afunc, shutdown=True)
-
- @unittest.skipIf(sys.platform == "win32", "curio does not work in windows CI")
- class CurioNoSniffioAsyncDetectionTests(NoSniffioAsyncDetectionTests):
- expect_raise = True
-
- def async_run(self, afunc):
- with curio.Kernel() as kernel:
- return kernel.run(afunc, shutdown=True)
-
- @unittest.skipIf(sys.platform == "win32", "curio does not work in windows CI")
- class CurioAsyncTests(AsyncTests):
- connect_udp = False
-
- def setUp(self):
- self.backend = dns.asyncbackend.set_default_backend("curio")
-
- def async_run(self, afunc):
- with curio.Kernel() as kernel:
- return kernel.run(afunc, shutdown=True)
-
-except ImportError:
- pass
except ImportError:
pass
-try:
- import curio
-
- @pytest.mark.skipif(
- (not _nanonameserver_available) or sys.platform == "win32",
- reason="requires nanonameserver or is windows",
- )
- def test_curio_inbound_xfr():
- dns.asyncbackend.set_default_backend("curio")
-
- async def run():
- await async_inbound_xfr()
-
- curio.run(run)
-
-except ImportError:
- pass
-
class UDPXFRNanoNameserver(Server):
def __init__(self):
projects / thirdparty / dnspython.git / commitdiff
