summaryrefslogtreecommitdiff
path: root/tests/test_doq.py
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