(path: Path, timeout_s: float = 10.0)
| 24 | |
| 25 | |
| 26 | def wait_for_socket(path: Path, timeout_s: float = 10.0) -> None: |
| 27 | deadline = time.time() + timeout_s |
| 28 | while time.time() < deadline: |
| 29 | if path.exists(): |
| 30 | try: |
| 31 | sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) |
| 32 | sock.settimeout(0.2) |
| 33 | sock.connect(str(path)) |
| 34 | sock.close() |
| 35 | return |
| 36 | except OSError: |
| 37 | pass |
| 38 | time.sleep(0.01) |
| 39 | raise RuntimeError(f"socket not ready: {path}") |
| 40 | |
| 41 | |
| 42 | def create_session(debug_sock: Path, cwd: str) -> str: |
no test coverage detected