diff --git a/src/openai/resources/beta/realtime/realtime.py b/src/openai/resources/beta/realtime/realtime.py index 07883f272b..ee70d639b3 100644 --- a/src/openai/resources/beta/realtime/realtime.py +++ b/src/openai/resources/beta/realtime/realtime.py @@ -44,7 +44,7 @@ TranscriptionSessionsWithStreamingResponse, AsyncTranscriptionSessionsWithStreamingResponse, ) -from ....types.websocket_connection_options import WebsocketConnectionOptions +from ....types.websocket_connection_options import WebsocketConnectionOptions, SyncWebSocketConnectionOptions from ....types.beta.realtime.realtime_client_event import RealtimeClientEvent from ....types.beta.realtime.realtime_server_event import RealtimeServerEvent from ....types.beta.realtime.conversation_item_param import ConversationItemParam @@ -95,7 +95,7 @@ def connect( model: str, extra_query: Query = {}, extra_headers: Headers = {}, - websocket_connection_options: WebsocketConnectionOptions = {}, + websocket_connection_options: SyncWebSocketConnectionOptions = {}, ) -> RealtimeConnectionManager: """ The Realtime API enables you to build low-latency, multi-modal conversational experiences. It currently supports text and audio as both input and output, as well as function calling. @@ -382,6 +382,7 @@ async def __aenter__(self) -> AsyncRealtimeConnection: if self.__websocket_connection_options: log.debug("Custom WebSocket connection options provided") + options: WebsocketConnectionOptions = {"max_size": None, **self.__websocket_connection_options} self.__connection = AsyncRealtimeConnection( await connect( str(url), @@ -393,7 +394,7 @@ async def __aenter__(self) -> AsyncRealtimeConnection: }, self.__extra_headers, ), - **self.__websocket_connection_options, + **options, ) ) @@ -524,7 +525,7 @@ def __init__( model: str, extra_query: Query, extra_headers: Headers, - websocket_connection_options: WebsocketConnectionOptions, + websocket_connection_options: SyncWebSocketConnectionOptions, ) -> None: self.__client = client self.__model = model @@ -575,6 +576,7 @@ def __enter__(self) -> RealtimeConnection: if self.__websocket_connection_options: log.debug("Custom WebSocket connection options provided") + options: SyncWebSocketConnectionOptions = {"max_size": None, **self.__websocket_connection_options} self.__connection = RealtimeConnection( connect( str(url), @@ -586,7 +588,7 @@ def __enter__(self) -> RealtimeConnection: }, self.__extra_headers, ), - **self.__websocket_connection_options, + **options, ) ) diff --git a/tests/lib/test_beta_realtime_websocket_options.py b/tests/lib/test_beta_realtime_websocket_options.py new file mode 100644 index 0000000000..259b9b43bd --- /dev/null +++ b/tests/lib/test_beta_realtime_websocket_options.py @@ -0,0 +1,84 @@ +import json +import asyncio +from typing import Literal, Iterator, Optional +from threading import Thread + +import pytest +from websockets.exceptions import ConnectionClosed, ConnectionClosedError +from websockets.sync.server import ServerConnection, serve + +from openai import OpenAI, AsyncOpenAI +from openai.types.websocket_connection_options import WebSocketConnectionOptions + + +@pytest.fixture(autouse=True) +def bypass_loopback_proxy(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("NO_PROXY", "*") + monkeypatch.setenv("no_proxy", "*") + + +@pytest.fixture +def realtime_peer(payload_size: int) -> Iterator[tuple[str, bytes]]: + wire = json.dumps({"type": "future.event", "text": "x" * payload_size}).encode() + + def accept(socket: ServerConnection) -> None: + socket.send(wire) + try: + socket.recv() + except ConnectionClosed: + # The client closes after consuming or rejecting the test event. + pass + + with serve(accept, "127.0.0.1", 0) as server: + thread = Thread(target=server.serve_forever) + thread.start() + try: + yield f"ws://127.0.0.1:{server.socket.getsockname()[1]}", wire + finally: + server.shutdown() + thread.join(3) + assert not thread.is_alive() + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize( + ("payload_size", "max_size", "reject"), + [ + pytest.param(128, "default", False, id="default-small"), + pytest.param(2**20 + 32, "default", False, id="default-over-native"), + pytest.param(2**20 + 32, None, False, id="explicit-unlimited"), + pytest.param(128, 512, False, id="explicit-limit-under"), + pytest.param(900, 512, True, id="explicit-limit-over"), + pytest.param(1, 0, True, id="explicit-zero"), + ], +) +async def test_beta_realtime_preserves_receive_limit_choice( + realtime_peer: tuple[str, bytes], mode: Literal["sync", "async"], max_size: Optional[int | str], reject: bool +) -> None: + url, expected = realtime_peer + options: WebSocketConnectionOptions = {} + if isinstance(max_size, int) or max_size is None: + options["max_size"] = max_size + + if mode == "sync": + + def receive() -> bytes: + with OpenAI(api_key="local", websocket_base_url=url) as client: + with client.beta.realtime.connect(model="test", websocket_connection_options=options) as conn: + return conn.recv_bytes() + + if reject: + with pytest.raises(ConnectionClosedError) as exc: + await asyncio.wait_for(asyncio.to_thread(receive), 10) + assert exc.value.sent is not None and exc.value.sent.code == 1009 + else: + assert await asyncio.wait_for(asyncio.to_thread(receive), 10) == expected + else: + async with AsyncOpenAI(api_key="local", websocket_base_url=url) as client: + async with client.beta.realtime.connect(model="test", websocket_connection_options=options) as conn: + if reject: + with pytest.raises(ConnectionClosedError) as exc: + await asyncio.wait_for(conn.recv_bytes(), 10) + assert exc.value.sent is not None and exc.value.sent.code == 1009 + else: + assert await asyncio.wait_for(conn.recv_bytes(), 10) == expected diff --git a/tests/test_websocket_connection_options.py b/tests/test_websocket_connection_options.py index 468b797c8a..9f95ce9672 100644 --- a/tests/test_websocket_connection_options.py +++ b/tests/test_websocket_connection_options.py @@ -25,7 +25,7 @@ def test_beta_realtime_import_still_works_with_old_alias() -> None: if TYPE_CHECKING: - from openai import AsyncOpenAI + from openai import OpenAI, AsyncOpenAI from openai.types.websocket_connection_options import SyncWebSocketConnectionOptions def native_options_contract( @@ -49,3 +49,8 @@ def native_options_contract( client.realtime.connect(model="test", websocket_connection_options=original) client.responses.connect(websocket_connection_options=legacy) return valid_sync, invalid_sync + + def beta_realtime_options_contract(client: OpenAI, async_client: AsyncOpenAI) -> None: + client.beta.realtime.connect(model="test", websocket_connection_options={"max_size": 512}) + client.beta.realtime.connect(model="test", websocket_connection_options={"write_limit": 65536}) # pyright: ignore[reportArgumentType] + async_client.beta.realtime.connect(model="test", websocket_connection_options={"write_limit": 65536})