graphql-engine/server/tests-py/ports.py

import contextlib
import socket
import time
from typing import Optional

def find_free_port() -> int:
    """
    Finds a free port.

    There is no lock placed on the port, so something else could claim the port
    between this function finding a port and returning.
    """
    with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
        s.bind(('', 0))
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        return s.getsockname()[1]

def is_port_in_use(port: int) -> bool:
    """
    Checks whether a local port is in use.
    """
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        return s.connect_ex(('localhost', port)) == 0

def wait_for_port(port: int, timeout: Optional[float] = None) -> None:
    """
    Waits until a port is opened, with an optional timeout.
    """
    start_time = time.monotonic()
    while timeout is None or (time.monotonic() - start_time) < timeout:
        if is_port_in_use(port):
            return
        time.sleep(0.2)
    raise TimeoutError(f'Timed out waiting for port {port}.')