GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

openai/openai-python

Public

mirrored fromhttps://github.com/openai/openai-pythonAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
dev/codex/pin-setup-rye-workflows-main

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

openai-python/tests/api_resources/realtime

tests/api_resources/realtime/test_calls.py

692lines · modecode

RawDownload
Latest commit unavailable.
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 httpx
9import pytest
10from respx import MockRouter
11
12import openai._legacy_response as _legacy_response
13from openai import OpenAI, AsyncOpenAI
14from tests.utils import assert_matches_type
15
16# pyright: reportDeprecated=false
17
18base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
19
20
21class TestCalls:
22 parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
23
24 @parametrize
25 @pytest.mark.respx(base_url=base_url)
26 def test_method_create(self, client: OpenAI, respx_mock: MockRouter) -> None:
27 respx_mock.post("/realtime/calls").mock(return_value=httpx.Response(200, json={"foo": "bar"}))
28 call = client.realtime.calls.create(
29 sdp="sdp",
30 )
31 assert isinstance(call, _legacy_response.HttpxBinaryResponseContent)
32 assert call.json() == {"foo": "bar"}
33
34 @parametrize
35 @pytest.mark.respx(base_url=base_url)
36 def test_method_create_with_all_params(self, client: OpenAI, respx_mock: MockRouter) -> None:
37 respx_mock.post("/realtime/calls").mock(return_value=httpx.Response(200, json={"foo": "bar"}))
38 call = client.realtime.calls.create(
39 sdp="sdp",
40 session={
41 "type": "realtime",
42 "audio": {
43 "input": {
44 "format": {
45 "rate": 24000,
46 "type": "audio/pcm",
47 },
48 "noise_reduction": {"type": "near_field"},
49 "transcription": {
50 "language": "language",
51 "model": "string",
52 "prompt": "prompt",
53 },
54 "turn_detection": {
55 "type": "server_vad",
56 "create_response": True,
57 "idle_timeout_ms": 5000,
58 "interrupt_response": True,
59 "prefix_padding_ms": 0,
60 "silence_duration_ms": 0,
61 "threshold": 0,
62 },
63 },
64 "output": {
65 "format": {
66 "rate": 24000,
67 "type": "audio/pcm",
68 },
69 "speed": 0.25,
70 "voice": "string",
71 },
72 },
73 "include": ["item.input_audio_transcription.logprobs"],
74 "instructions": "instructions",
75 "max_output_tokens": 0,
76 "model": "string",
77 "output_modalities": ["text"],
78 "prompt": {
79 "id": "id",
80 "variables": {"foo": "string"},
81 "version": "version",
82 },
83 "tool_choice": "none",
84 "tools": [
85 {
86 "description": "description",
87 "name": "name",
88 "parameters": {},
89 "type": "function",
90 }
91 ],
92 "tracing": "auto",
93 "truncation": "auto",
94 },
95 )
96 assert isinstance(call, _legacy_response.HttpxBinaryResponseContent)
97 assert call.json() == {"foo": "bar"}
98
99 @parametrize
100 @pytest.mark.respx(base_url=base_url)
101 def test_raw_response_create(self, client: OpenAI, respx_mock: MockRouter) -> None:
102 respx_mock.post("/realtime/calls").mock(return_value=httpx.Response(200, json={"foo": "bar"}))
103
104 response = client.realtime.calls.with_raw_response.create(
105 sdp="sdp",
106 )
107
108 assert response.is_closed is True
109 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
110 call = response.parse()
111 assert_matches_type(_legacy_response.HttpxBinaryResponseContent, call, path=["response"])
112
113 @parametrize
114 @pytest.mark.respx(base_url=base_url)
115 def test_streaming_response_create(self, client: OpenAI, respx_mock: MockRouter) -> None:
116 respx_mock.post("/realtime/calls").mock(return_value=httpx.Response(200, json={"foo": "bar"}))
117 with client.realtime.calls.with_streaming_response.create(
118 sdp="sdp",
119 ) as response:
120 assert not response.is_closed
121 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
122
123 call = response.parse()
124 assert_matches_type(bytes, call, path=["response"])
125
126 assert cast(Any, response.is_closed) is True
127
128 @parametrize
129 def test_method_accept(self, client: OpenAI) -> None:
130 call = client.realtime.calls.accept(
131 call_id="call_id",
132 type="realtime",
133 )
134 assert call is None
135
136 @parametrize
137 def test_method_accept_with_all_params(self, client: OpenAI) -> None:
138 call = client.realtime.calls.accept(
139 call_id="call_id",
140 type="realtime",
141 audio={
142 "input": {
143 "format": {
144 "rate": 24000,
145 "type": "audio/pcm",
146 },
147 "noise_reduction": {"type": "near_field"},
148 "transcription": {
149 "language": "language",
150 "model": "string",
151 "prompt": "prompt",
152 },
153 "turn_detection": {
154 "type": "server_vad",
155 "create_response": True,
156 "idle_timeout_ms": 5000,
157 "interrupt_response": True,
158 "prefix_padding_ms": 0,
159 "silence_duration_ms": 0,
160 "threshold": 0,
161 },
162 },
163 "output": {
164 "format": {
165 "rate": 24000,
166 "type": "audio/pcm",
167 },
168 "speed": 0.25,
169 "voice": "string",
170 },
171 },
172 include=["item.input_audio_transcription.logprobs"],
173 instructions="instructions",
174 max_output_tokens=0,
175 model="string",
176 output_modalities=["text"],
177 prompt={
178 "id": "id",
179 "variables": {"foo": "string"},
180 "version": "version",
181 },
182 tool_choice="none",
183 tools=[
184 {
185 "description": "description",
186 "name": "name",
187 "parameters": {},
188 "type": "function",
189 }
190 ],
191 tracing="auto",
192 truncation="auto",
193 )
194 assert call is None
195
196 @parametrize
197 def test_raw_response_accept(self, client: OpenAI) -> None:
198 response = client.realtime.calls.with_raw_response.accept(
199 call_id="call_id",
200 type="realtime",
201 )
202
203 assert response.is_closed is True
204 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
205 call = response.parse()
206 assert call is None
207
208 @parametrize
209 def test_streaming_response_accept(self, client: OpenAI) -> None:
210 with client.realtime.calls.with_streaming_response.accept(
211 call_id="call_id",
212 type="realtime",
213 ) as response:
214 assert not response.is_closed
215 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
216
217 call = response.parse()
218 assert call is None
219
220 assert cast(Any, response.is_closed) is True
221
222 @parametrize
223 def test_path_params_accept(self, client: OpenAI) -> None:
224 with pytest.raises(ValueError, match=r"Expected a non-empty value for `call_id` but received ''"):
225 client.realtime.calls.with_raw_response.accept(
226 call_id="",
227 type="realtime",
228 )
229
230 @parametrize
231 def test_method_hangup(self, client: OpenAI) -> None:
232 call = client.realtime.calls.hangup(
233 "call_id",
234 )
235 assert call is None
236
237 @parametrize
238 def test_raw_response_hangup(self, client: OpenAI) -> None:
239 response = client.realtime.calls.with_raw_response.hangup(
240 "call_id",
241 )
242
243 assert response.is_closed is True
244 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
245 call = response.parse()
246 assert call is None
247
248 @parametrize
249 def test_streaming_response_hangup(self, client: OpenAI) -> None:
250 with client.realtime.calls.with_streaming_response.hangup(
251 "call_id",
252 ) as response:
253 assert not response.is_closed
254 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
255
256 call = response.parse()
257 assert call is None
258
259 assert cast(Any, response.is_closed) is True
260
261 @parametrize
262 def test_path_params_hangup(self, client: OpenAI) -> None:
263 with pytest.raises(ValueError, match=r"Expected a non-empty value for `call_id` but received ''"):
264 client.realtime.calls.with_raw_response.hangup(
265 "",
266 )
267
268 @parametrize
269 def test_method_refer(self, client: OpenAI) -> None:
270 call = client.realtime.calls.refer(
271 call_id="call_id",
272 target_uri="tel:+14155550123",
273 )
274 assert call is None
275
276 @parametrize
277 def test_raw_response_refer(self, client: OpenAI) -> None:
278 response = client.realtime.calls.with_raw_response.refer(
279 call_id="call_id",
280 target_uri="tel:+14155550123",
281 )
282
283 assert response.is_closed is True
284 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
285 call = response.parse()
286 assert call is None
287
288 @parametrize
289 def test_streaming_response_refer(self, client: OpenAI) -> None:
290 with client.realtime.calls.with_streaming_response.refer(
291 call_id="call_id",
292 target_uri="tel:+14155550123",
293 ) as response:
294 assert not response.is_closed
295 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
296
297 call = response.parse()
298 assert call is None
299
300 assert cast(Any, response.is_closed) is True
301
302 @parametrize
303 def test_path_params_refer(self, client: OpenAI) -> None:
304 with pytest.raises(ValueError, match=r"Expected a non-empty value for `call_id` but received ''"):
305 client.realtime.calls.with_raw_response.refer(
306 call_id="",
307 target_uri="tel:+14155550123",
308 )
309
310 @parametrize
311 def test_method_reject(self, client: OpenAI) -> None:
312 call = client.realtime.calls.reject(
313 call_id="call_id",
314 )
315 assert call is None
316
317 @parametrize
318 def test_method_reject_with_all_params(self, client: OpenAI) -> None:
319 call = client.realtime.calls.reject(
320 call_id="call_id",
321 status_code=486,
322 )
323 assert call is None
324
325 @parametrize
326 def test_raw_response_reject(self, client: OpenAI) -> None:
327 response = client.realtime.calls.with_raw_response.reject(
328 call_id="call_id",
329 )
330
331 assert response.is_closed is True
332 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
333 call = response.parse()
334 assert call is None
335
336 @parametrize
337 def test_streaming_response_reject(self, client: OpenAI) -> None:
338 with client.realtime.calls.with_streaming_response.reject(
339 call_id="call_id",
340 ) as response:
341 assert not response.is_closed
342 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
343
344 call = response.parse()
345 assert call is None
346
347 assert cast(Any, response.is_closed) is True
348
349 @parametrize
350 def test_path_params_reject(self, client: OpenAI) -> None:
351 with pytest.raises(ValueError, match=r"Expected a non-empty value for `call_id` but received ''"):
352 client.realtime.calls.with_raw_response.reject(
353 call_id="",
354 )
355
356
357class TestAsyncCalls:
358 parametrize = pytest.mark.parametrize(
359 "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
360 )
361
362 @parametrize
363 @pytest.mark.respx(base_url=base_url)
364 async def test_method_create(self, async_client: AsyncOpenAI, respx_mock: MockRouter) -> None:
365 respx_mock.post("/realtime/calls").mock(return_value=httpx.Response(200, json={"foo": "bar"}))
366 call = await async_client.realtime.calls.create(
367 sdp="sdp",
368 )
369 assert isinstance(call, _legacy_response.HttpxBinaryResponseContent)
370 assert call.json() == {"foo": "bar"}
371
372 @parametrize
373 @pytest.mark.respx(base_url=base_url)
374 async def test_method_create_with_all_params(self, async_client: AsyncOpenAI, respx_mock: MockRouter) -> None:
375 respx_mock.post("/realtime/calls").mock(return_value=httpx.Response(200, json={"foo": "bar"}))
376 call = await async_client.realtime.calls.create(
377 sdp="sdp",
378 session={
379 "type": "realtime",
380 "audio": {
381 "input": {
382 "format": {
383 "rate": 24000,
384 "type": "audio/pcm",
385 },
386 "noise_reduction": {"type": "near_field"},
387 "transcription": {
388 "language": "language",
389 "model": "string",
390 "prompt": "prompt",
391 },
392 "turn_detection": {
393 "type": "server_vad",
394 "create_response": True,
395 "idle_timeout_ms": 5000,
396 "interrupt_response": True,
397 "prefix_padding_ms": 0,
398 "silence_duration_ms": 0,
399 "threshold": 0,
400 },
401 },
402 "output": {
403 "format": {
404 "rate": 24000,
405 "type": "audio/pcm",
406 },
407 "speed": 0.25,
408 "voice": "string",
409 },
410 },
411 "include": ["item.input_audio_transcription.logprobs"],
412 "instructions": "instructions",
413 "max_output_tokens": 0,
414 "model": "string",
415 "output_modalities": ["text"],
416 "prompt": {
417 "id": "id",
418 "variables": {"foo": "string"},
419 "version": "version",
420 },
421 "tool_choice": "none",
422 "tools": [
423 {
424 "description": "description",
425 "name": "name",
426 "parameters": {},
427 "type": "function",
428 }
429 ],
430 "tracing": "auto",
431 "truncation": "auto",
432 },
433 )
434 assert isinstance(call, _legacy_response.HttpxBinaryResponseContent)
435 assert call.json() == {"foo": "bar"}
436
437 @parametrize
438 @pytest.mark.respx(base_url=base_url)
439 async def test_raw_response_create(self, async_client: AsyncOpenAI, respx_mock: MockRouter) -> None:
440 respx_mock.post("/realtime/calls").mock(return_value=httpx.Response(200, json={"foo": "bar"}))
441
442 response = await async_client.realtime.calls.with_raw_response.create(
443 sdp="sdp",
444 )
445
446 assert response.is_closed is True
447 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
448 call = response.parse()
449 assert_matches_type(_legacy_response.HttpxBinaryResponseContent, call, path=["response"])
450
451 @parametrize
452 @pytest.mark.respx(base_url=base_url)
453 async def test_streaming_response_create(self, async_client: AsyncOpenAI, respx_mock: MockRouter) -> None:
454 respx_mock.post("/realtime/calls").mock(return_value=httpx.Response(200, json={"foo": "bar"}))
455 async with async_client.realtime.calls.with_streaming_response.create(
456 sdp="sdp",
457 ) as response:
458 assert not response.is_closed
459 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
460
461 call = await response.parse()
462 assert_matches_type(bytes, call, path=["response"])
463
464 assert cast(Any, response.is_closed) is True
465
466 @parametrize
467 async def test_method_accept(self, async_client: AsyncOpenAI) -> None:
468 call = await async_client.realtime.calls.accept(
469 call_id="call_id",
470 type="realtime",
471 )
472 assert call is None
473
474 @parametrize
475 async def test_method_accept_with_all_params(self, async_client: AsyncOpenAI) -> None:
476 call = await async_client.realtime.calls.accept(
477 call_id="call_id",
478 type="realtime",
479 audio={
480 "input": {
481 "format": {
482 "rate": 24000,
483 "type": "audio/pcm",
484 },
485 "noise_reduction": {"type": "near_field"},
486 "transcription": {
487 "language": "language",
488 "model": "string",
489 "prompt": "prompt",
490 },
491 "turn_detection": {
492 "type": "server_vad",
493 "create_response": True,
494 "idle_timeout_ms": 5000,
495 "interrupt_response": True,
496 "prefix_padding_ms": 0,
497 "silence_duration_ms": 0,
498 "threshold": 0,
499 },
500 },
501 "output": {
502 "format": {
503 "rate": 24000,
504 "type": "audio/pcm",
505 },
506 "speed": 0.25,
507 "voice": "string",
508 },
509 },
510 include=["item.input_audio_transcription.logprobs"],
511 instructions="instructions",
512 max_output_tokens=0,
513 model="string",
514 output_modalities=["text"],
515 prompt={
516 "id": "id",
517 "variables": {"foo": "string"},
518 "version": "version",
519 },
520 tool_choice="none",
521 tools=[
522 {
523 "description": "description",
524 "name": "name",
525 "parameters": {},
526 "type": "function",
527 }
528 ],
529 tracing="auto",
530 truncation="auto",
531 )
532 assert call is None
533
534 @parametrize
535 async def test_raw_response_accept(self, async_client: AsyncOpenAI) -> None:
536 response = await async_client.realtime.calls.with_raw_response.accept(
537 call_id="call_id",
538 type="realtime",
539 )
540
541 assert response.is_closed is True
542 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
543 call = response.parse()
544 assert call is None
545
546 @parametrize
547 async def test_streaming_response_accept(self, async_client: AsyncOpenAI) -> None:
548 async with async_client.realtime.calls.with_streaming_response.accept(
549 call_id="call_id",
550 type="realtime",
551 ) as response:
552 assert not response.is_closed
553 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
554
555 call = await response.parse()
556 assert call is None
557
558 assert cast(Any, response.is_closed) is True
559
560 @parametrize
561 async def test_path_params_accept(self, async_client: AsyncOpenAI) -> None:
562 with pytest.raises(ValueError, match=r"Expected a non-empty value for `call_id` but received ''"):
563 await async_client.realtime.calls.with_raw_response.accept(
564 call_id="",
565 type="realtime",
566 )
567
568 @parametrize
569 async def test_method_hangup(self, async_client: AsyncOpenAI) -> None:
570 call = await async_client.realtime.calls.hangup(
571 "call_id",
572 )
573 assert call is None
574
575 @parametrize
576 async def test_raw_response_hangup(self, async_client: AsyncOpenAI) -> None:
577 response = await async_client.realtime.calls.with_raw_response.hangup(
578 "call_id",
579 )
580
581 assert response.is_closed is True
582 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
583 call = response.parse()
584 assert call is None
585
586 @parametrize
587 async def test_streaming_response_hangup(self, async_client: AsyncOpenAI) -> None:
588 async with async_client.realtime.calls.with_streaming_response.hangup(
589 "call_id",
590 ) as response:
591 assert not response.is_closed
592 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
593
594 call = await response.parse()
595 assert call is None
596
597 assert cast(Any, response.is_closed) is True
598
599 @parametrize
600 async def test_path_params_hangup(self, async_client: AsyncOpenAI) -> None:
601 with pytest.raises(ValueError, match=r"Expected a non-empty value for `call_id` but received ''"):
602 await async_client.realtime.calls.with_raw_response.hangup(
603 "",
604 )
605
606 @parametrize
607 async def test_method_refer(self, async_client: AsyncOpenAI) -> None:
608 call = await async_client.realtime.calls.refer(
609 call_id="call_id",
610 target_uri="tel:+14155550123",
611 )
612 assert call is None
613
614 @parametrize
615 async def test_raw_response_refer(self, async_client: AsyncOpenAI) -> None:
616 response = await async_client.realtime.calls.with_raw_response.refer(
617 call_id="call_id",
618 target_uri="tel:+14155550123",
619 )
620
621 assert response.is_closed is True
622 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
623 call = response.parse()
624 assert call is None
625
626 @parametrize
627 async def test_streaming_response_refer(self, async_client: AsyncOpenAI) -> None:
628 async with async_client.realtime.calls.with_streaming_response.refer(
629 call_id="call_id",
630 target_uri="tel:+14155550123",
631 ) as response:
632 assert not response.is_closed
633 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
634
635 call = await response.parse()
636 assert call is None
637
638 assert cast(Any, response.is_closed) is True
639
640 @parametrize
641 async def test_path_params_refer(self, async_client: AsyncOpenAI) -> None:
642 with pytest.raises(ValueError, match=r"Expected a non-empty value for `call_id` but received ''"):
643 await async_client.realtime.calls.with_raw_response.refer(
644 call_id="",
645 target_uri="tel:+14155550123",
646 )
647
648 @parametrize
649 async def test_method_reject(self, async_client: AsyncOpenAI) -> None:
650 call = await async_client.realtime.calls.reject(
651 call_id="call_id",
652 )
653 assert call is None
654
655 @parametrize
656 async def test_method_reject_with_all_params(self, async_client: AsyncOpenAI) -> None:
657 call = await async_client.realtime.calls.reject(
658 call_id="call_id",
659 status_code=486,
660 )
661 assert call is None
662
663 @parametrize
664 async def test_raw_response_reject(self, async_client: AsyncOpenAI) -> None:
665 response = await async_client.realtime.calls.with_raw_response.reject(
666 call_id="call_id",
667 )
668
669 assert response.is_closed is True
670 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
671 call = response.parse()
672 assert call is None
673
674 @parametrize
675 async def test_streaming_response_reject(self, async_client: AsyncOpenAI) -> None:
676 async with async_client.realtime.calls.with_streaming_response.reject(
677 call_id="call_id",
678 ) as response:
679 assert not response.is_closed
680 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
681
682 call = await response.parse()
683 assert call is None
684
685 assert cast(Any, response.is_closed) is True
686
687 @parametrize
688 async def test_path_params_reject(self, async_client: AsyncOpenAI) -> None:
689 with pytest.raises(ValueError, match=r"Expected a non-empty value for `call_id` but received ''"):
690 await async_client.realtime.calls.with_raw_response.reject(
691 call_id="",
692 )
693