| 310 | |
| 311 | |
| 312 | def _build_websocket_request( |
| 313 | port: int, |
| 314 | path: str = "/ws", |
| 315 | headers: dict[str, str] | None = None, |
| 316 | ) -> bytes: |
| 317 | key = base64.b64encode(os.urandom(16)).decode("ascii") |
| 318 | request_headers = { |
| 319 | "Host": f"127.0.0.1:{port}", |
| 320 | "Upgrade": "websocket", |
| 321 | "Connection": "Upgrade", |
| 322 | "Sec-WebSocket-Key": key, |
| 323 | "Sec-WebSocket-Version": "13", |
| 324 | **(headers or {}), |
| 325 | } |
| 326 | |
| 327 | request_lines = [f"GET {path} HTTP/1.1"] |
| 328 | request_lines.extend( |
| 329 | f"{name}: {value}" for name, value in request_headers.items() |
| 330 | ) |
| 331 | return ("\r\n".join([*request_lines, "", ""])).encode("ascii") |
| 332 | |
| 333 | |
| 334 | class _WebSocketClient: |