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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
"""
HTTP server fixture for tests which binds to an auto-assigned port on localhost.
"""
import asyncio
import contextlib
import dataclasses
import socket
import threading
import time
from queue import Queue
from typing import Tuple
import aiohttp.web as web
@dataclasses.dataclass
class HttpServer:
app: web.Application
port: int
class Event_ts(asyncio.Event):
"""
A thread safe version of the asyncio Event
NOTE: clear() is not thread safe
Taken from https://stackoverflow.com/a/33006667
"""
def __init__(self, *args, loop: asyncio.AbstractEventLoop | None = None, **kwargs):
"""
Creates a thread-safe event for the given loop (or the loop of the current thread).
"""
super().__init__(*args, **kwargs)
self.target_loop = loop or asyncio.get_running_loop()
def set(self):
self.target_loop.call_soon_threadsafe(super().set)
def _make_localhost_socket() -> Tuple[socket.socket, int]:
"""Creates a localhost-bound socket with an auto-assigned port."""
sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
sock.bind(("::1", 0))
# Shouldn't matter because we dynamically allocate ports, but this is generally preferred.
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
_, port = sock.getsockname()[:2]
return (sock, port)
def _server_thread(app: web.Application, sock: socket.socket, shutdown_ev_q: Queue):
async def async_main():
nonlocal app, sock
# Due to Reasons(tm) of event loop lifecycles and stuff of the sort,
# it's far easier to just send the event object to the other thread
# from inside the loop where it already knows which loop it is.
shutdown_ev = Event_ts()
shutdown_ev_q.put(shutdown_ev)
runner = web.AppRunner(app, handle_signals=False)
await runner.setup()
site = web.SockSite(runner, sock)
await site.start()
await shutdown_ev.wait()
await runner.cleanup()
asyncio.run(async_main())
@contextlib.contextmanager
def http_server(app: web.Application):
"""
Creates an http server on an automatically chosen port on the host
running the given web.Application, gives you the port for it.
The server is run on a separate thread.
"""
# n.b. pytest doesn't directly support asyncio. There's a bunch of
# complexity that we could go through to do this or we could just throw the
# async on a thread which was what we would do to the web server anyway if
# it was blocking.
shutdown_ev_q = Queue()
thr = None
sock = None
shutdown_ev = None
try:
sock, port = _make_localhost_socket()
thr = threading.Thread(
target=_server_thread,
args=(app, sock, shutdown_ev_q),
name=f"functional2 httpd [::1]:{port}",
)
thr.start()
shutdown_ev = shutdown_ev_q.get()
yield HttpServer(app=app, port=port)
finally:
if shutdown_ev:
shutdown_ev.set()
if thr:
thr.join()
if sock:
sock.close()
def dev_main():
"""A little test server for poking at this manually"""
async def root(_req: web.Request):
return web.Response(body="hello world")
app = web.Application()
app.add_routes([web.get("/", root)])
with http_server(app) as httpd:
print(f"Listening on http://[::1]:{httpd.port}")
time.sleep(3600)
if __name__ == "__main__":
dev_main()
|
