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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
|
from __future__ import annotations
import asyncio
import ssl
import sys
from collections.abc import Generator
from ssl import SSLContext
from typing import Any
from unittest.mock import Mock
import pytest
import trustme
from _pytest.fixtures import SubRequest
from trustme import CA
uvloop_marks = []
try:
import uvloop
except ImportError:
uvloop_marks.append(pytest.mark.skip(reason="uvloop not available"))
uvloop = Mock()
else:
if hasattr(asyncio.AbstractEventLoop, "shutdown_default_executor") and not hasattr(
uvloop.loop.Loop, "shutdown_default_executor"
):
uvloop_marks.append(
pytest.mark.skip(reason="uvloop is missing shutdown_default_executor()")
)
pytest_plugins = ["pytester"]
asyncio_params = [
pytest.param(("asyncio", {"debug": True}), id="asyncio"),
pytest.param(
("asyncio", {"debug": True, "loop_factory": uvloop.new_event_loop}),
marks=uvloop_marks,
id="asyncio+uvloop",
),
]
if sys.version_info >= (3, 12):
def eager_task_loop_factory() -> asyncio.AbstractEventLoop:
loop = asyncio.new_event_loop()
loop.set_task_factory(asyncio.eager_task_factory)
return loop
asyncio_params.append(
pytest.param(
("asyncio", {"debug": True, "loop_factory": eager_task_loop_factory}),
id="asyncio+eager",
),
)
@pytest.fixture(params=[*asyncio_params, pytest.param("trio")])
def anyio_backend(request: SubRequest) -> tuple[str, dict[str, Any]]:
return request.param
@pytest.fixture(scope="session")
def ca() -> CA:
return trustme.CA()
@pytest.fixture(scope="session")
def server_context(ca: CA) -> SSLContext:
server_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
if hasattr(ssl, "OP_IGNORE_UNEXPECTED_EOF"):
server_context.options &= ~ssl.OP_IGNORE_UNEXPECTED_EOF
ca.issue_cert("localhost").configure_cert(server_context)
return server_context
@pytest.fixture(scope="session")
def client_context(ca: CA) -> SSLContext:
client_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
if hasattr(ssl, "OP_IGNORE_UNEXPECTED_EOF"):
client_context.options &= ~ssl.OP_IGNORE_UNEXPECTED_EOF
ca.configure_trust(client_context)
return client_context
@pytest.fixture
def asyncio_event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
loop = asyncio.DefaultEventLoopPolicy().new_event_loop()
asyncio.set_event_loop(loop)
yield loop
asyncio.set_event_loop(None)
loop.close()
|
