import socket from collections.abc import Mapping def _get_random_port() -> int: # os automatically assigns an available port when binding to port 0, so we can use that to find a free port try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(('', 0)) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) port = s.getsockname()[1] return port except Exception as e: raise RuntimeError(f"Could not find an available port: {e}") def _get_service_port(mapping: str | int | Mapping[str, object]) -> int | None: if isinstance(mapping, int): return mapping if isinstance(mapping, Mapping): published = mapping.get("published") if published is None: return None try: return int(published) except (TypeError, ValueError): return None parts = str(mapping).split(":") match parts: case [""]: return None case [port] | [port, _]: try: return int(port) except ValueError: return None case ["localhost" | "127.0.0.1", _, _]: return None case [_, port, _]: try: return int(port) except ValueError: return None case _: return None def _replace_port_in_mapping( mapping: str | int | Mapping[str, object], port_from: int, port_to: int, ) -> str | dict[str, object]: if isinstance(mapping, Mapping): updated = dict(mapping) if _get_service_port(updated) == port_from: updated["published"] = str(port_to) return updated parts = str(mapping).split(":") match parts: case [_]: return str(port_to) case [_, container_port]: return f"{port_to}:{container_port}" case [ip, _, container_port]: return f"{ip}:{port_to}:{container_port}" case _: return str(port_to) def _get_port_index(mappings: list[str | int | Mapping[str, object]], port: int) -> int | None: for i, mapping in enumerate(mappings): if _get_service_port(mapping) == port: return i return None