0
Self Samples
750
Cumulative
1
Lines Hit
2.58%
% of Total
0
Max Self
750
Max Total
Intensity:
Cold
Hot
Self Time
Total Time
Show All
Hot Only
Heat
Specialization
Line
Self
Total
Code
1
"""Run the unmodified PyPI app on a temporary loopback port."""
2
3
import argparse
4
import asyncio
5
import json
6
import os
7
import socket
8
import time
9
from pathlib import Path
10
11
import uvicorn
12
from web_edit.app import create_app
13
14
15
async def serve(directory: Path, state: Path, timeout: float):
16
# Binding port zero avoids port-selection races with other local servers.
17
with socket.socket() as sock:
18
sock.bind(("127.0.0.1", 0))
19
server = uvicorn.Server(
20
uvicorn.Config(
21
create_app(directory),
22
log_level="warning",
23
access_log=False,
24
lifespan="off",
25
)
26
)
27
task = asyncio.create_task(server.serve(sockets=[sock]))
28
try:
29
while not server.started:
30
if task.done():
31
await task
32
raise RuntimeError("Server exited before becoming ready")
33
await asyncio.sleep(0.02)
34
ready = state / "ready.json"
35
temporary = state / "ready.tmp"
36
temporary.write_text(
37
json.dumps({"pid": os.getpid(), "port": sock.getsockname()[1]})
38
)
39
temporary.replace(ready)
40
deadline = time.monotonic() + timeout
41
while not (state / "stop").exists() and time.monotonic() < deadline:
42
if task.done():
43
await task
44
return
45
await asyncio.sleep(0.1)
46
finally:
47
server.should_exit = True
48
await task
49
50
51
if __name__ == "__main__":
52
parser = argparse.ArgumentParser(description=__doc__)
53
parser.add_argument("directory", type=Path)
54
parser.add_argument("state", type=Path)
55
parser.add_argument("--timeout", type=float, default=120)
56
args = parser.parse_args()
57
750
asyncio.run(serve(args.directory, args.state, args.timeout))