36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import socket
|
|
from collections.abc import Iterable
|
|
|
|
|
|
def pick_free_port(host: str = "127.0.0.1", candidates: Iterable[int] | None = None) -> int:
|
|
"""Choose an available TCP port without binding a privileged/system port."""
|
|
|
|
ports = list(candidates or range(49152, 65536))
|
|
for port in ports:
|
|
with socket.socket(
|
|
socket.AF_INET6 if ":" in host else socket.AF_INET, socket.SOCK_STREAM
|
|
) as sock:
|
|
try:
|
|
sock.bind((host, port))
|
|
except OSError:
|
|
continue
|
|
return port
|
|
raise OSError("no free high TCP port")
|
|
|
|
|
|
async def drain_process_output(process: asyncio.subprocess.Process) -> list[str]:
|
|
"""Read both streams until a process exits, returning sanitized text lines."""
|
|
|
|
lines: list[str] = []
|
|
|
|
async def read(stream: asyncio.StreamReader | None) -> None:
|
|
if stream is None:
|
|
return
|
|
while line := await stream.readline():
|
|
lines.append(line.decode("utf-8", errors="replace").rstrip())
|
|
|
|
await asyncio.gather(read(process.stdout), read(process.stderr))
|
|
return lines
|