From b64d42ebb228171433a7f128b561f086198f930d Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 00:01:56 +0000 Subject: [PATCH 1/2] fix: preserve beta Realtime byte reads on websockets 13 --- .../resources/beta/realtime/realtime.py | 4 +- tests/lib/test_realtime_websocket_contract.py | 15 +- tests/lib/test_websocket_redirects.py | 133 +++++++++++++++++- 3 files changed, 146 insertions(+), 6 deletions(-) diff --git a/src/openai/resources/beta/realtime/realtime.py b/src/openai/resources/beta/realtime/realtime.py index 99f3d10191..07883f272b 100644 --- a/src/openai/resources/beta/realtime/realtime.py +++ b/src/openai/resources/beta/realtime/realtime.py @@ -469,7 +469,9 @@ def recv_bytes(self) -> bytes: If you want to parse the message into a `RealtimeServerEvent` object like `.recv()` does, then you can call `.parse_event(data)`. """ - message = self._connection.recv(decode=False) + from ....lib._websocket import _recv_bytes + + message = _recv_bytes(self._connection) log.debug("Received WebSocket message: %i bytes", len(message)) return message diff --git a/tests/lib/test_realtime_websocket_contract.py b/tests/lib/test_realtime_websocket_contract.py index 005c4f88c4..63aaaf2599 100644 --- a/tests/lib/test_realtime_websocket_contract.py +++ b/tests/lib/test_realtime_websocket_contract.py @@ -139,8 +139,11 @@ def script(socket: ServerConnection) -> None: } +@pytest.mark.parametrize("beta", [False, True], ids=["realtime", "beta-realtime"]) @pytest.mark.parametrize("legacy_recv", [False, True], ids=["native", "websockets-13-signature"]) -def test_realtime_recv_bytes_preserves_text_and_binary(monkeypatch: pytest.MonkeyPatch, legacy_recv: bool) -> None: +def test_realtime_recv_bytes_preserves_text_and_binary( + monkeypatch: pytest.MonkeyPatch, legacy_recv: bool, beta: bool +) -> None: if legacy_recv: original_recv = ClientConnection.recv @@ -158,8 +161,14 @@ def script(socket: ServerConnection) -> None: socket.send(frame) with script_server(script) as url: - with OpenAI(api_key="synthetic", base_url=url, http_client=httpx2.Client(trust_env=False)) as client: - with client.realtime.connect(model="gpt-realtime") as connection: + with OpenAI( + api_key="synthetic", + base_url=url, + websocket_base_url=url.replace("http://", "ws://"), + http_client=httpx2.Client(trust_env=False), + ) as client: + resource = client.beta.realtime if beta else client.realtime + with resource.connect(model="gpt-realtime") as connection: for frame in frames: expected = frame.encode("utf-8") if isinstance(frame, str) else frame received = connection.recv_bytes() diff --git a/tests/lib/test_websocket_redirects.py b/tests/lib/test_websocket_redirects.py index 3515744ef8..1727a8aec0 100644 --- a/tests/lib/test_websocket_redirects.py +++ b/tests/lib/test_websocket_redirects.py @@ -2,14 +2,17 @@ import asyncio import inspect -from typing import Any, Callable +from queue import Queue +from typing import Any, Callable, Iterator +from contextlib import AbstractContextManager from unittest.mock import Mock, AsyncMock import httpx2 import pytest from websockets.uri import WebSocketURI -from websockets.http11 import Response +from websockets.http11 import Request, Response from websockets.exceptions import InvalidStatus, SecurityError +from websockets.sync.server import ServerConnection from websockets.asyncio.client import connect from websockets.datastructures import Headers, HeadersLike @@ -237,3 +240,129 @@ def test_sync_websocket_connector_is_unchanged(monkeypatch: pytest.MonkeyPatch, def test_non_redirect_error_is_preserved() -> None: error = InvalidStatus(Response(401, "Unauthorized", Headers())) assert _WebSocketConnect("wss://origin.test/realtime").process_redirect(error) is error + + +WirePeer = Callable[[str | None], AbstractContextManager[tuple[str, list[Request], Queue[bool]]]] + + +@pytest.fixture +def wire_peer(monkeypatch: pytest.MonkeyPatch) -> WirePeer: + """Use real sockets for both the synchronous and asynchronous clients.""" + import threading + from contextlib import contextmanager + + from websockets.sync.server import serve + + monkeypatch.setenv("NO_PROXY", "*") + monkeypatch.setenv("no_proxy", "*") + + @contextmanager + def start(redirect: str | None = None) -> Iterator[tuple[str, list[Request], Queue[bool]]]: + requests: list[Request] = [] + closed: Queue[bool] = Queue() + + def process(_socket: ServerConnection, request: Request) -> Response | None: + requests.append(request) + if redirect and not request.path.startswith("/final"): + return Response(302, "Found", Headers({"Location": redirect})) + return None + + def echo(socket: ServerConnection) -> None: + try: + for message in socket: + socket.send(message) + finally: + closed.put(True) + + with serve(echo, "127.0.0.1", 0, process_request=process) as server: + thread = threading.Thread(target=server.serve_forever) + thread.start() + try: + yield f"ws://127.0.0.1:{server.socket.getsockname()[1]}", requests, closed + finally: + server.shutdown() + thread.join(timeout=5) + assert not thread.is_alive() + + return start + + +@pytest.mark.parametrize("name", RESOURCES) +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("redirect", ["none", "same-origin", "cross-origin"]) +async def test_public_binding_wire_redirects(wire_peer: WirePeer, name: str, mode: str, redirect: str) -> None: + from websockets.exceptions import InvalidHandshake + + with wire_peer(None) as (target, target_requests, _): + location = {"none": None, "same-origin": "/final", "cross-origin": target + "/final"}[redirect] + with wire_peer(location) as (base, requests, closed): + allowed = redirect == "none" or (redirect == "same-origin" and mode == "async" and FOLLOWS_REDIRECTS) + + async def run() -> None: + if mode == "sync": + with OpenAI(api_key="fake-wire-key", websocket_base_url=base) as client: + with resource(client, name).connect(**options(client, name)) as connection: + connection.send({"type": "future.event"}) + assert connection.recv().type == "future.event" + else: + async with AsyncOpenAI(api_key="fake-wire-key", websocket_base_url=base) as client: + async with resource(client, name).connect(**options(client, name)) as connection: + await connection.send({"type": "future.event"}) + assert (await connection.recv()).type == "future.event" + + if allowed: + await run() + else: + with pytest.raises(InvalidHandshake): + await run() + assert len(requests) == (2 if allowed and redirect == "same-origin" else 1) + if allowed: + assert closed.get(timeout=5) is True + assert closed.empty() + assert target_requests == [] + for request in requests: + assert request.headers["Authorization"] == "Bearer fake-wire-key" + for key, value in EXTRA_HEADERS.items(): + assert request.headers[key] == value + + +@pytest.mark.parametrize("name", RESOURCES) +@pytest.mark.parametrize("mode", ["sync", "async"]) +async def test_public_binding_wire_credentials(wire_peer: WirePeer, name: str, mode: str) -> None: + calls = 0 + + def credential() -> str: + nonlocal calls + calls += 1 + return f"fake-wire-{calls}" + + async def async_credential() -> str: + return credential() + + with wire_peer(None) as (base, requests, closed): + if mode == "sync": + with OpenAI(api_key=credential, websocket_base_url=base) as client: + for attempt in range(3): + kwargs = options(client, name) + if attempt == 2: + kwargs["extra_headers"] = {"Authorization": "Bearer fake-explicit"} + with resource(client, name).connect(**kwargs) as connection: + connection.send({"type": "future.event"}) + assert connection.recv().type == "future.event" + else: + async with AsyncOpenAI(api_key=async_credential, websocket_base_url=base) as client: + for attempt in range(3): + kwargs = options(client, name) + if attempt == 2: + kwargs["extra_headers"] = {"Authorization": "Bearer fake-explicit"} + async with resource(client, name).connect(**kwargs) as connection: + await connection.send({"type": "future.event"}) + assert (await connection.recv()).type == "future.event" + assert [request.headers["Authorization"] for request in requests] == [ + "Bearer fake-wire-1", + "Bearer fake-wire-2", + "Bearer fake-explicit", + ] + for _ in range(3): + assert closed.get(timeout=5) is True + assert closed.empty() From f6ca372f5d5153448f5f641d1fc7d73794a09b27 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 00:03:29 +0000 Subject: [PATCH 2/2] test: use generator annotation for WebSocket fixture --- tests/lib/test_websocket_redirects.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/lib/test_websocket_redirects.py b/tests/lib/test_websocket_redirects.py index 1727a8aec0..90668cd61b 100644 --- a/tests/lib/test_websocket_redirects.py +++ b/tests/lib/test_websocket_redirects.py @@ -3,7 +3,7 @@ import asyncio import inspect from queue import Queue -from typing import Any, Callable, Iterator +from typing import Any, Callable, Generator from contextlib import AbstractContextManager from unittest.mock import Mock, AsyncMock @@ -257,7 +257,7 @@ def wire_peer(monkeypatch: pytest.MonkeyPatch) -> WirePeer: monkeypatch.setenv("no_proxy", "*") @contextmanager - def start(redirect: str | None = None) -> Iterator[tuple[str, list[Request], Queue[bool]]]: + def start(redirect: str | None = None) -> Generator[tuple[str, list[Request], Queue[bool]], None, None]: requests: list[Request] = [] closed: Queue[bool] = Queue()