Jade Lovelace
c57a99e9c3
This doesn't use http.server because it is really quite janky and has a quite bad API. cc: https://git.lix.systems/lix-project/lix/issues/600 Change-Id: I48b7c1beb89fac21fa32f1e09f595063b795d70a
125 lines
3.5 KiB
Python
125 lines
3.5 KiB
Python
"""
|
|
HTTP server fixture for tests which binds to an auto-assigned port on localhost.
|
|
"""
|
|
import asyncio
|
|
import contextlib
|
|
import dataclasses
|
|
import time
|
|
import socket
|
|
import threading
|
|
from typing import Tuple
|
|
from queue import Queue
|
|
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()
|