main
1# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
2
3from __future__ import annotations
4
5import os
6from typing import Any, cast
7
8import pytest
9
10from openai import OpenAI, AsyncOpenAI
11from tests.utils import assert_matches_type
12from openai.types.realtime import ClientSecretCreateResponse
13
14base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
15
16
17class TestClientSecrets:
18 parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
19
20 @parametrize
21 def test_method_create(self, client: OpenAI) -> None:
22 client_secret = client.realtime.client_secrets.create()
23 assert_matches_type(ClientSecretCreateResponse, client_secret, path=["response"])
24
25 @parametrize
26 def test_method_create_with_all_params(self, client: OpenAI) -> None:
27 client_secret = client.realtime.client_secrets.create(
28 expires_after={
29 "anchor": "created_at",
30 "seconds": 10,
31 },
32 session={
33 "type": "realtime",
34 "audio": {
35 "input": {
36 "format": {
37 "rate": 24000,
38 "type": "audio/pcm",
39 },
40 "noise_reduction": {"type": "near_field"},
41 "transcription": {
42 "language": "language",
43 "model": "whisper-1",
44 "prompt": "prompt",
45 },
46 "turn_detection": {
47 "type": "server_vad",
48 "create_response": True,
49 "idle_timeout_ms": 5000,
50 "interrupt_response": True,
51 "prefix_padding_ms": 0,
52 "silence_duration_ms": 0,
53 "threshold": 0,
54 },
55 },
56 "output": {
57 "format": {
58 "rate": 24000,
59 "type": "audio/pcm",
60 },
61 "speed": 0.25,
62 "voice": "ash",
63 },
64 },
65 "include": ["item.input_audio_transcription.logprobs"],
66 "instructions": "instructions",
67 "max_output_tokens": 0,
68 "model": "string",
69 "output_modalities": ["text"],
70 "prompt": {
71 "id": "id",
72 "variables": {"foo": "string"},
73 "version": "version",
74 },
75 "tool_choice": "none",
76 "tools": [
77 {
78 "description": "description",
79 "name": "name",
80 "parameters": {},
81 "type": "function",
82 }
83 ],
84 "tracing": "auto",
85 "truncation": "auto",
86 },
87 )
88 assert_matches_type(ClientSecretCreateResponse, client_secret, path=["response"])
89
90 @parametrize
91 def test_raw_response_create(self, client: OpenAI) -> None:
92 response = client.realtime.client_secrets.with_raw_response.create()
93
94 assert response.is_closed is True
95 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
96 client_secret = response.parse()
97 assert_matches_type(ClientSecretCreateResponse, client_secret, path=["response"])
98
99 @parametrize
100 def test_streaming_response_create(self, client: OpenAI) -> None:
101 with client.realtime.client_secrets.with_streaming_response.create() as response:
102 assert not response.is_closed
103 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
104
105 client_secret = response.parse()
106 assert_matches_type(ClientSecretCreateResponse, client_secret, path=["response"])
107
108 assert cast(Any, response.is_closed) is True
109
110
111class TestAsyncClientSecrets:
112 parametrize = pytest.mark.parametrize(
113 "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
114 )
115
116 @parametrize
117 async def test_method_create(self, async_client: AsyncOpenAI) -> None:
118 client_secret = await async_client.realtime.client_secrets.create()
119 assert_matches_type(ClientSecretCreateResponse, client_secret, path=["response"])
120
121 @parametrize
122 async def test_method_create_with_all_params(self, async_client: AsyncOpenAI) -> None:
123 client_secret = await async_client.realtime.client_secrets.create(
124 expires_after={
125 "anchor": "created_at",
126 "seconds": 10,
127 },
128 session={
129 "type": "realtime",
130 "audio": {
131 "input": {
132 "format": {
133 "rate": 24000,
134 "type": "audio/pcm",
135 },
136 "noise_reduction": {"type": "near_field"},
137 "transcription": {
138 "language": "language",
139 "model": "whisper-1",
140 "prompt": "prompt",
141 },
142 "turn_detection": {
143 "type": "server_vad",
144 "create_response": True,
145 "idle_timeout_ms": 5000,
146 "interrupt_response": True,
147 "prefix_padding_ms": 0,
148 "silence_duration_ms": 0,
149 "threshold": 0,
150 },
151 },
152 "output": {
153 "format": {
154 "rate": 24000,
155 "type": "audio/pcm",
156 },
157 "speed": 0.25,
158 "voice": "ash",
159 },
160 },
161 "include": ["item.input_audio_transcription.logprobs"],
162 "instructions": "instructions",
163 "max_output_tokens": 0,
164 "model": "string",
165 "output_modalities": ["text"],
166 "prompt": {
167 "id": "id",
168 "variables": {"foo": "string"},
169 "version": "version",
170 },
171 "tool_choice": "none",
172 "tools": [
173 {
174 "description": "description",
175 "name": "name",
176 "parameters": {},
177 "type": "function",
178 }
179 ],
180 "tracing": "auto",
181 "truncation": "auto",
182 },
183 )
184 assert_matches_type(ClientSecretCreateResponse, client_secret, path=["response"])
185
186 @parametrize
187 async def test_raw_response_create(self, async_client: AsyncOpenAI) -> None:
188 response = await async_client.realtime.client_secrets.with_raw_response.create()
189
190 assert response.is_closed is True
191 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
192 client_secret = response.parse()
193 assert_matches_type(ClientSecretCreateResponse, client_secret, path=["response"])
194
195 @parametrize
196 async def test_streaming_response_create(self, async_client: AsyncOpenAI) -> None:
197 async with async_client.realtime.client_secrets.with_streaming_response.create() as response:
198 assert not response.is_closed
199 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
200
201 client_secret = await response.parse()
202 assert_matches_type(ClientSecretCreateResponse, client_secret, path=["response"])
203
204 assert cast(Any, response.is_closed) is True