import socket from collections.abc import Mapping def get_random_port() -> int: # Binding to port 0 asks the OS for an available ephemeral 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}") from e def _get_service_binding( mapping: str | int | Mapping[str, object], ) -> tuple[int | None, str | None]: if isinstance(mapping, int): return mapping, None if isinstance(mapping, Mapping): published = mapping.get("published") if published is None: return None, None try: port = int(published) except TypeError: return None, None except ValueError: return None, None host = mapping.get("host_ip") return port, str(host) if host not in {None, ""} else None parts = str(mapping).split(":") match parts: case [""]: return None, None case [port] | [port, _]: try: return int(port), None except ValueError: return None, None case [host, port, _]: try: return int(port), host or None except ValueError: return None, None case _: return None, None def _get_service_port(mapping: str | int | Mapping[str, object]) -> int | None: port, _host = _get_service_binding(mapping) return port def _replace_port_in_mapping( mapping: str | int | Mapping[str, object], port_from: int, port_to: int, host_to: str | None = None, rewrite_host: bool = False, ) -> str | dict[str, object]: if isinstance(mapping, Mapping): updated = dict(mapping) if _get_service_port(updated) == port_from: updated["published"] = str(port_to) if rewrite_host: if host_to is None: updated.pop("host_ip", None) else: updated["host_ip"] = host_to return updated parts = str(mapping).split(":") match parts: case [_]: return str(port_to) case [_, container_port]: if rewrite_host and host_to: return f"{host_to}:{port_to}:{container_port}" return f"{port_to}:{container_port}" case [ip, _, container_port]: if rewrite_host: if host_to: return f"{host_to}:{port_to}:{container_port}" return f"{port_to}:{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