(path: str, timeout_s: float)
| 115 | |
| 116 | |
| 117 | def wait_for_socket(path: str, timeout_s: float) -> bool: |
| 118 | deadline = time.perf_counter() + timeout_s |
| 119 | while time.perf_counter() < deadline: |
| 120 | if os.path.exists(path): |
| 121 | try: |
| 122 | sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) |
| 123 | sock.connect(path) |
| 124 | sock.close() |
| 125 | return True |
| 126 | except OSError: |
| 127 | pass |
| 128 | time.sleep(0.005) |
| 129 | return False |
| 130 | |
| 131 | |
| 132 | def measure_server_startup(binary: str, runs: int) -> list[float]: |
no test coverage detected