Bind the socket to a free port and return the port number. Relies on ephemeral ports in order to ensure we are using an unbound port. This is important as many tests may be running simultaneously, especially in a buildbot environment. This method raises an exception if the sock.family
(sock: socket.socket, host: str = HOST)
| 76 | |
| 77 | |
| 78 | def bind_port(sock: socket.socket, host: str = HOST) -> int: |
| 79 | """Bind the socket to a free port and return the port number. Relies on |
| 80 | ephemeral ports in order to ensure we are using an unbound port. This is |
| 81 | important as many tests may be running simultaneously, especially in a |
| 82 | buildbot environment. This method raises an exception if the sock.family |
| 83 | is AF_INET and sock.type is SOCK_STREAM, *and* the socket has SO_REUSEADDR |
| 84 | or SO_REUSEPORT set on it. Tests should *never* set these socket options |
| 85 | for TCP/IP sockets. The only case for setting these options is testing |
| 86 | multicasting via multiple UDP sockets. |
| 87 | |
| 88 | Additionally, if the SO_EXCLUSIVEADDRUSE socket option is available (i.e. |
| 89 | on Windows), it will be set on the socket. This will prevent anyone else |
| 90 | from bind()'ing to our host/port for the duration of the test. |
| 91 | """ |
| 92 | |
| 93 | if sock.family == socket.AF_INET and sock.type == socket.SOCK_STREAM: |
| 94 | if hasattr(socket, "SO_REUSEADDR"): |
| 95 | if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) == 1: |
| 96 | raise ValueError( |
| 97 | "tests should never set the SO_REUSEADDR " |
| 98 | "socket option on TCP/IP sockets!" |
| 99 | ) |
| 100 | if hasattr(socket, "SO_REUSEPORT"): |
| 101 | try: |
| 102 | if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT) == 1: |
| 103 | raise ValueError( |
| 104 | "tests should never set the SO_REUSEPORT " |
| 105 | "socket option on TCP/IP sockets!" |
| 106 | ) |
| 107 | except OSError: |
| 108 | # Python's socket module was compiled using modern headers |
| 109 | # thus defining SO_REUSEPORT but this process is running |
| 110 | # under an older kernel that does not support SO_REUSEPORT. |
| 111 | pass |
| 112 | if hasattr(socket, "SO_EXCLUSIVEADDRUSE"): |
| 113 | sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1) |
| 114 | |
| 115 | sock.bind((host, 0)) |
| 116 | port = sock.getsockname()[1] |
| 117 | assert isinstance(port, int) |
| 118 | return port |