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
v1.14.0

Branches

Tags

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

Clone

HTTPS

Download ZIP

openai-python/tests/api_resources/beta/threads/runs

tests/api_resources/beta/threads/runs/test_steps.py

263lines · modecode

RawDownload
Latest commit unavailable.
1# File generated from our OpenAPI spec by Stainless.
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.beta.threads.runs import RunStep
14
15base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
16
17
18class TestSteps:
19 parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
20
21 @parametrize
22 def test_method_retrieve(self, client: OpenAI) -> None:
23 step = client.beta.threads.runs.steps.retrieve(
24 "string",
25 thread_id="string",
26 run_id="string",
27 )
28 assert_matches_type(RunStep, step, path=["response"])
29
30 @parametrize
31 def test_raw_response_retrieve(self, client: OpenAI) -> None:
32 response = client.beta.threads.runs.steps.with_raw_response.retrieve(
33 "string",
34 thread_id="string",
35 run_id="string",
36 )
37
38 assert response.is_closed is True
39 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
40 step = response.parse()
41 assert_matches_type(RunStep, step, path=["response"])
42
43 @parametrize
44 def test_streaming_response_retrieve(self, client: OpenAI) -> None:
45 with client.beta.threads.runs.steps.with_streaming_response.retrieve(
46 "string",
47 thread_id="string",
48 run_id="string",
49 ) as response:
50 assert not response.is_closed
51 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
52
53 step = response.parse()
54 assert_matches_type(RunStep, step, path=["response"])
55
56 assert cast(Any, response.is_closed) is True
57
58 @parametrize
59 def test_path_params_retrieve(self, client: OpenAI) -> None:
60 with pytest.raises(ValueError, match=r"Expected a non-empty value for `thread_id` but received ''"):
61 client.beta.threads.runs.steps.with_raw_response.retrieve(
62 "string",
63 thread_id="",
64 run_id="string",
65 )
66
67 with pytest.raises(ValueError, match=r"Expected a non-empty value for `run_id` but received ''"):
68 client.beta.threads.runs.steps.with_raw_response.retrieve(
69 "string",
70 thread_id="string",
71 run_id="",
72 )
73
74 with pytest.raises(ValueError, match=r"Expected a non-empty value for `step_id` but received ''"):
75 client.beta.threads.runs.steps.with_raw_response.retrieve(
76 "",
77 thread_id="string",
78 run_id="string",
79 )
80
81 @parametrize
82 def test_method_list(self, client: OpenAI) -> None:
83 step = client.beta.threads.runs.steps.list(
84 "string",
85 thread_id="string",
86 )
87 assert_matches_type(SyncCursorPage[RunStep], step, path=["response"])
88
89 @parametrize
90 def test_method_list_with_all_params(self, client: OpenAI) -> None:
91 step = client.beta.threads.runs.steps.list(
92 "string",
93 thread_id="string",
94 after="string",
95 before="string",
96 limit=0,
97 order="asc",
98 )
99 assert_matches_type(SyncCursorPage[RunStep], step, path=["response"])
100
101 @parametrize
102 def test_raw_response_list(self, client: OpenAI) -> None:
103 response = client.beta.threads.runs.steps.with_raw_response.list(
104 "string",
105 thread_id="string",
106 )
107
108 assert response.is_closed is True
109 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
110 step = response.parse()
111 assert_matches_type(SyncCursorPage[RunStep], step, path=["response"])
112
113 @parametrize
114 def test_streaming_response_list(self, client: OpenAI) -> None:
115 with client.beta.threads.runs.steps.with_streaming_response.list(
116 "string",
117 thread_id="string",
118 ) as response:
119 assert not response.is_closed
120 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
121
122 step = response.parse()
123 assert_matches_type(SyncCursorPage[RunStep], step, path=["response"])
124
125 assert cast(Any, response.is_closed) is True
126
127 @parametrize
128 def test_path_params_list(self, client: OpenAI) -> None:
129 with pytest.raises(ValueError, match=r"Expected a non-empty value for `thread_id` but received ''"):
130 client.beta.threads.runs.steps.with_raw_response.list(
131 "string",
132 thread_id="",
133 )
134
135 with pytest.raises(ValueError, match=r"Expected a non-empty value for `run_id` but received ''"):
136 client.beta.threads.runs.steps.with_raw_response.list(
137 "",
138 thread_id="string",
139 )
140
141
142class TestAsyncSteps:
143 parametrize = pytest.mark.parametrize("async_client", [False, True], indirect=True, ids=["loose", "strict"])
144
145 @parametrize
146 async def test_method_retrieve(self, async_client: AsyncOpenAI) -> None:
147 step = await async_client.beta.threads.runs.steps.retrieve(
148 "string",
149 thread_id="string",
150 run_id="string",
151 )
152 assert_matches_type(RunStep, step, path=["response"])
153
154 @parametrize
155 async def test_raw_response_retrieve(self, async_client: AsyncOpenAI) -> None:
156 response = await async_client.beta.threads.runs.steps.with_raw_response.retrieve(
157 "string",
158 thread_id="string",
159 run_id="string",
160 )
161
162 assert response.is_closed is True
163 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
164 step = response.parse()
165 assert_matches_type(RunStep, step, path=["response"])
166
167 @parametrize
168 async def test_streaming_response_retrieve(self, async_client: AsyncOpenAI) -> None:
169 async with async_client.beta.threads.runs.steps.with_streaming_response.retrieve(
170 "string",
171 thread_id="string",
172 run_id="string",
173 ) as response:
174 assert not response.is_closed
175 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
176
177 step = await response.parse()
178 assert_matches_type(RunStep, step, path=["response"])
179
180 assert cast(Any, response.is_closed) is True
181
182 @parametrize
183 async def test_path_params_retrieve(self, async_client: AsyncOpenAI) -> None:
184 with pytest.raises(ValueError, match=r"Expected a non-empty value for `thread_id` but received ''"):
185 await async_client.beta.threads.runs.steps.with_raw_response.retrieve(
186 "string",
187 thread_id="",
188 run_id="string",
189 )
190
191 with pytest.raises(ValueError, match=r"Expected a non-empty value for `run_id` but received ''"):
192 await async_client.beta.threads.runs.steps.with_raw_response.retrieve(
193 "string",
194 thread_id="string",
195 run_id="",
196 )
197
198 with pytest.raises(ValueError, match=r"Expected a non-empty value for `step_id` but received ''"):
199 await async_client.beta.threads.runs.steps.with_raw_response.retrieve(
200 "",
201 thread_id="string",
202 run_id="string",
203 )
204
205 @parametrize
206 async def test_method_list(self, async_client: AsyncOpenAI) -> None:
207 step = await async_client.beta.threads.runs.steps.list(
208 "string",
209 thread_id="string",
210 )
211 assert_matches_type(AsyncCursorPage[RunStep], step, path=["response"])
212
213 @parametrize
214 async def test_method_list_with_all_params(self, async_client: AsyncOpenAI) -> None:
215 step = await async_client.beta.threads.runs.steps.list(
216 "string",
217 thread_id="string",
218 after="string",
219 before="string",
220 limit=0,
221 order="asc",
222 )
223 assert_matches_type(AsyncCursorPage[RunStep], step, path=["response"])
224
225 @parametrize
226 async def test_raw_response_list(self, async_client: AsyncOpenAI) -> None:
227 response = await async_client.beta.threads.runs.steps.with_raw_response.list(
228 "string",
229 thread_id="string",
230 )
231
232 assert response.is_closed is True
233 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
234 step = response.parse()
235 assert_matches_type(AsyncCursorPage[RunStep], step, path=["response"])
236
237 @parametrize
238 async def test_streaming_response_list(self, async_client: AsyncOpenAI) -> None:
239 async with async_client.beta.threads.runs.steps.with_streaming_response.list(
240 "string",
241 thread_id="string",
242 ) as response:
243 assert not response.is_closed
244 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
245
246 step = await response.parse()
247 assert_matches_type(AsyncCursorPage[RunStep], step, path=["response"])
248
249 assert cast(Any, response.is_closed) is True
250
251 @parametrize
252 async def test_path_params_list(self, async_client: AsyncOpenAI) -> None:
253 with pytest.raises(ValueError, match=r"Expected a non-empty value for `thread_id` but received ''"):
254 await async_client.beta.threads.runs.steps.with_raw_response.list(
255 "string",
256 thread_id="",
257 )
258
259 with pytest.raises(ValueError, match=r"Expected a non-empty value for `run_id` but received ''"):
260 await async_client.beta.threads.runs.steps.with_raw_response.list(
261 "",
262 thread_id="string",
263 )
264