Method connect

python/vercel-runtime/tests/test_runtime.py:340–363

(
        cls,
        port: int,
        path: str = "/ws",
        headers: dict[str, str] | None = None,
    )

    @classmethod
    def connect(
        cls,
        port: int,
        path: str = "/ws",
        headers: dict[str, str] | None = None,
    ) -> tuple[_WebSocketClient, str]:
        sock = socket.create_connection(("127.0.0.1", port), timeout=5)
        sock.settimeout(5)
        reader = _SocketBuffer(sock)

        try:
            sock.sendall(_build_websocket_request(port, path, headers))

            response = reader.read_until(b"\r\n\r\n")
            if b"sec-websocket-accept:" not in response.lower():
                raise AssertionError(
                    "websocket upgrade response missing accept header"
                )

            status_line = response.split(b"\r\n", maxsplit=1)[0].decode("ascii")
            return cls(sock, reader), status_line
        except Exception:
            sock.close()
            raise

    def read_text(self) -> str:
        opcode, payload = _read_ws_frame(self._reader)