Returns a server port number that can be safely used for listening without clashing with another test worker process, when running with pytest-xdist. If multiple test workers invoke this function with the same min value, each of them will receive a different number that is not lower tha
(max_retries=10)
| 18 | used_ports = set() |
| 19 | |
| 20 | def get_test_server_port(max_retries=10): |
| 21 | """Returns a server port number that can be safely used for listening without |
| 22 | clashing with another test worker process, when running with pytest-xdist. |
| 23 | |
| 24 | If multiple test workers invoke this function with the same min value, each of |
| 25 | them will receive a different number that is not lower than start (but may be |
| 26 | higher). If the resulting value is >=stop, it is a fatal error. |
| 27 | |
| 28 | Note that if multiple test workers invoke this function with different ranges |
| 29 | that overlap, conflicts are possible! |
| 30 | |
| 31 | Args: |
| 32 | max_retries: Number of times to retry finding an available port |
| 33 | """ |
| 34 | |
| 35 | try: |
| 36 | worker_id = util.force_ascii(os.environ["PYTEST_XDIST_WORKER"]) |
| 37 | except KeyError: |
| 38 | n = 0 |
| 39 | else: |
| 40 | assert worker_id == some.bytes.matching( |
| 41 | rb"gw(\d+)" |
| 42 | ), "Unrecognized PYTEST_XDIST_WORKER format" |
| 43 | n = int(worker_id[2:]) |
| 44 | |
| 45 | # Try multiple times to find an available port, with retry logic |
| 46 | for attempt in range(max_retries): |
| 47 | port = 5678 + (n * 300) + attempt |
| 48 | while port in used_ports: |
| 49 | port += 1 |
| 50 | |
| 51 | # Verify the port is actually available by trying to bind to it |
| 52 | sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 53 | sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 54 | try: |
| 55 | sock.bind(("127.0.0.1", port)) |
| 56 | sock.close() |
| 57 | used_ports.add(port) |
| 58 | log.info("Allocated port {0} for worker {1}", port, n) |
| 59 | return port |
| 60 | except OSError as e: |
| 61 | log.warning("Port {0} unavailable (attempt {1}/{2}): {3}", port, attempt + 1, max_retries, e) |
| 62 | sock.close() |
| 63 | time.sleep(0.1 * (attempt + 1)) # Exponential backoff |
| 64 | |
| 65 | # Fall back to original behavior if all retries fail |
| 66 | port = 5678 + (n * 300) |
| 67 | while port in used_ports: |
| 68 | port += 1 |
| 69 | used_ports.add(port) |
| 70 | log.warning("Using fallback port {0} after {1} retries", port, max_retries) |
| 71 | return port |
| 72 | |
| 73 | |
| 74 | def find_http_url(text): |