openai/openai-python

Public

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

CodeCommitsIssuesPull requestsActionsInsightsSecurity
v1.100.0

Branches

Tags

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

Clone

HTTPS

Download ZIP

tests/api_resources/responses/test_input_items.py

125lines · modecode

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.pagination import SyncCursorPage, AsyncCursorPage
13from openai.types.responses import ResponseItem
14
15base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
16
17
18class TestInputItems:
19 parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
20
21 @parametrize
22 def test_method_list(self, client: OpenAI) -> None:
23 input_item = client.responses.input_items.list(
24 response_id="response_id",
25 )
26 assert_matches_type(SyncCursorPage[ResponseItem], input_item, path=["response"])
27
28 @parametrize
29 def test_method_list_with_all_params(self, client: OpenAI) -> None:
30 input_item = client.responses.input_items.list(
31 response_id="response_id",
32 after="after",
33 before="before",
34 include=["code_interpreter_call.outputs"],
35 limit=0,
36 order="asc",
37 )
38 assert_matches_type(SyncCursorPage[ResponseItem], input_item, path=["response"])
39
40 @parametrize
41 def test_raw_response_list(self, client: OpenAI) -> None:
42 response = client.responses.input_items.with_raw_response.list(
43 response_id="response_id",
44 )
45
46 assert response.is_closed is True
47 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
48 input_item = response.parse()
49 assert_matches_type(SyncCursorPage[ResponseItem], input_item, path=["response"])
50
51 @parametrize
52 def test_streaming_response_list(self, client: OpenAI) -> None:
53 with client.responses.input_items.with_streaming_response.list(
54 response_id="response_id",
55 ) as response:
56 assert not response.is_closed
57 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
58
59 input_item = response.parse()
60 assert_matches_type(SyncCursorPage[ResponseItem], input_item, path=["response"])
61
62 assert cast(Any, response.is_closed) is True
63
64 @parametrize
65 def test_path_params_list(self, client: OpenAI) -> None:
66 with pytest.raises(ValueError, match=r"Expected a non-empty value for `response_id` but received ''"):
67 client.responses.input_items.with_raw_response.list(
68 response_id="",
69 )
70
71
72class TestAsyncInputItems:
73 parametrize = pytest.mark.parametrize(
74 "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
75 )
76
77 @parametrize
78 async def test_method_list(self, async_client: AsyncOpenAI) -> None:
79 input_item = await async_client.responses.input_items.list(
80 response_id="response_id",
81 )
82 assert_matches_type(AsyncCursorPage[ResponseItem], input_item, path=["response"])
83
84 @parametrize
85 async def test_method_list_with_all_params(self, async_client: AsyncOpenAI) -> None:
86 input_item = await async_client.responses.input_items.list(
87 response_id="response_id",
88 after="after",
89 before="before",
90 include=["code_interpreter_call.outputs"],
91 limit=0,
92 order="asc",
93 )
94 assert_matches_type(AsyncCursorPage[ResponseItem], input_item, path=["response"])
95
96 @parametrize
97 async def test_raw_response_list(self, async_client: AsyncOpenAI) -> None:
98 response = await async_client.responses.input_items.with_raw_response.list(
99 response_id="response_id",
100 )
101
102 assert response.is_closed is True
103 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
104 input_item = response.parse()
105 assert_matches_type(AsyncCursorPage[ResponseItem], input_item, path=["response"])
106
107 @parametrize
108 async def test_streaming_response_list(self, async_client: AsyncOpenAI) -> None:
109 async with async_client.responses.input_items.with_streaming_response.list(
110 response_id="response_id",
111 ) as response:
112 assert not response.is_closed
113 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
114
115 input_item = await response.parse()
116 assert_matches_type(AsyncCursorPage[ResponseItem], input_item, path=["response"])
117
118 assert cast(Any, response.is_closed) is True
119
120 @parametrize
121 async def test_path_params_list(self, async_client: AsyncOpenAI) -> None:
122 with pytest.raises(ValueError, match=r"Expected a non-empty value for `response_id` but received ''"):
123 await async_client.responses.input_items.with_raw_response.list(
124 response_id="",
125 )
126