generated from mononoki/pyproj
42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
"""Execute the real entrypoint with a private TCP binding and graceful test shutdown.
|
|
|
|
Only asyncio.start_server's binding arguments are redirected. Its real server,
|
|
TCP.run, handler, manager, router, protocol and background sends all run normally.
|
|
The stdin task cancels main just as an external stop would, allowing coverage's
|
|
atexit hook to save child measurements on Windows as well as Linux.
|
|
"""
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
import runpy
|
|
import sys
|
|
|
|
|
|
async def stop_from_stdin(main_task):
|
|
await asyncio.to_thread(sys.stdin.buffer.readline)
|
|
main_task.cancel()
|
|
|
|
|
|
def run():
|
|
main = Path(sys.argv[1]).resolve()
|
|
original_start_server = asyncio.start_server
|
|
|
|
async def start_server(callback, host=None, port=None, **kwargs):
|
|
server = await original_start_server(callback, "127.0.0.1", 0, **kwargs)
|
|
address = server.sockets[0].getsockname()
|
|
asyncio.create_task(stop_from_stdin(asyncio.current_task()))
|
|
print(f"TCP listening {address[0]}:{address[1]}", flush=True)
|
|
return server
|
|
|
|
asyncio.start_server = start_server
|
|
sys.path.insert(0, str(main.parent))
|
|
sys.argv = [str(main)]
|
|
try:
|
|
runpy.run_path(str(main), run_name="__main__")
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|