blob: 254b578ed2f8cbfbecb11914272ea3990b9bb80c (
plain)
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
|
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
import asyncio
import sys
import pytest
import dns.asyncbackend
import dns.asyncquery
import dns.message
import dns.query
import dns.rcode
from .util import here
try:
from .nanoquic import Server
_nanoquic_available = True
except ImportError:
_nanoquic_available = False
class Server(object):
pass
@pytest.mark.skipif(not _nanoquic_available, reason="requires nanoquic")
def test_basic_sync():
with Server() as server:
q = dns.message.make_query("www.example.", "A")
r = dns.query.quic(q, "127.0.0.1", port=8853, verify=here("tls/ca.crt"))
assert r.rcode() == dns.rcode.REFUSED
async def amain():
q = dns.message.make_query("www.example.", "A")
r = await dns.asyncquery.quic(q, "127.0.0.1", port=8853, verify=here("tls/ca.crt"))
assert r.rcode() == dns.rcode.REFUSED
@pytest.mark.skipif(not _nanoquic_available, reason="requires nanoquic")
def test_basic_asyncio():
dns.asyncbackend.set_default_backend("asyncio")
with Server() as server:
asyncio.run(amain())
try:
import trio
@pytest.mark.skipif(not _nanoquic_available, reason="requires nanoquic")
def test_basic_trio():
dns.asyncbackend.set_default_backend("trio")
with Server() as server:
trio.run(amain)
except ImportError:
pass
|