openai/openai-python

Public

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

CodeCommitsIssuesPull requestsActionsInsightsSecurity
e9c26d1e253ff54ea50ea1b11d87c35405b24dda

Branches

Tags

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

Clone

HTTPS

Download ZIP

tests/api_resources/beta/test_threads.py

318lines · modecode

1# File generated from our OpenAPI spec by Stainless.
2
3from __future__ import annotations
4
5import os
6
7import pytest
8
9from openai import OpenAI, AsyncOpenAI
10from tests.utils import assert_matches_type
11from openai._client import OpenAI, AsyncOpenAI
12from openai.types.beta import Thread, ThreadDeleted
13from openai.types.beta.threads import Run
14
15base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
16api_key = "My API Key"
17
18
19class TestThreads:
20 strict_client = OpenAI(base_url=base_url, api_key=api_key, _strict_response_validation=True)
21 loose_client = OpenAI(base_url=base_url, api_key=api_key, _strict_response_validation=False)
22 parametrize = pytest.mark.parametrize("client", [strict_client, loose_client], ids=["strict", "loose"])
23
24 @parametrize
25 def test_method_create(self, client: OpenAI) -> None:
26 thread = client.beta.threads.create()
27 assert_matches_type(Thread, thread, path=["response"])
28
29 @parametrize
30 def test_method_create_with_all_params(self, client: OpenAI) -> None:
31 thread = client.beta.threads.create(
32 messages=[
33 {
34 "role": "user",
35 "content": "x",
36 "file_ids": ["string"],
37 "metadata": {},
38 },
39 {
40 "role": "user",
41 "content": "x",
42 "file_ids": ["string"],
43 "metadata": {},
44 },
45 {
46 "role": "user",
47 "content": "x",
48 "file_ids": ["string"],
49 "metadata": {},
50 },
51 ],
52 metadata={},
53 )
54 assert_matches_type(Thread, thread, path=["response"])
55
56 @parametrize
57 def test_raw_response_create(self, client: OpenAI) -> None:
58 response = client.beta.threads.with_raw_response.create()
59 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
60 thread = response.parse()
61 assert_matches_type(Thread, thread, path=["response"])
62
63 @parametrize
64 def test_method_retrieve(self, client: OpenAI) -> None:
65 thread = client.beta.threads.retrieve(
66 "string",
67 )
68 assert_matches_type(Thread, thread, path=["response"])
69
70 @parametrize
71 def test_raw_response_retrieve(self, client: OpenAI) -> None:
72 response = client.beta.threads.with_raw_response.retrieve(
73 "string",
74 )
75 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
76 thread = response.parse()
77 assert_matches_type(Thread, thread, path=["response"])
78
79 @parametrize
80 def test_method_update(self, client: OpenAI) -> None:
81 thread = client.beta.threads.update(
82 "string",
83 )
84 assert_matches_type(Thread, thread, path=["response"])
85
86 @parametrize
87 def test_method_update_with_all_params(self, client: OpenAI) -> None:
88 thread = client.beta.threads.update(
89 "string",
90 metadata={},
91 )
92 assert_matches_type(Thread, thread, path=["response"])
93
94 @parametrize
95 def test_raw_response_update(self, client: OpenAI) -> None:
96 response = client.beta.threads.with_raw_response.update(
97 "string",
98 )
99 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
100 thread = response.parse()
101 assert_matches_type(Thread, thread, path=["response"])
102
103 @parametrize
104 def test_method_delete(self, client: OpenAI) -> None:
105 thread = client.beta.threads.delete(
106 "string",
107 )
108 assert_matches_type(ThreadDeleted, thread, path=["response"])
109
110 @parametrize
111 def test_raw_response_delete(self, client: OpenAI) -> None:
112 response = client.beta.threads.with_raw_response.delete(
113 "string",
114 )
115 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
116 thread = response.parse()
117 assert_matches_type(ThreadDeleted, thread, path=["response"])
118
119 @parametrize
120 def test_method_create_and_run(self, client: OpenAI) -> None:
121 thread = client.beta.threads.create_and_run(
122 assistant_id="string",
123 )
124 assert_matches_type(Run, thread, path=["response"])
125
126 @parametrize
127 def test_method_create_and_run_with_all_params(self, client: OpenAI) -> None:
128 thread = client.beta.threads.create_and_run(
129 assistant_id="string",
130 instructions="string",
131 metadata={},
132 model="string",
133 thread={
134 "messages": [
135 {
136 "role": "user",
137 "content": "x",
138 "file_ids": ["string"],
139 "metadata": {},
140 },
141 {
142 "role": "user",
143 "content": "x",
144 "file_ids": ["string"],
145 "metadata": {},
146 },
147 {
148 "role": "user",
149 "content": "x",
150 "file_ids": ["string"],
151 "metadata": {},
152 },
153 ],
154 "metadata": {},
155 },
156 tools=[{"type": "code_interpreter"}, {"type": "code_interpreter"}, {"type": "code_interpreter"}],
157 )
158 assert_matches_type(Run, thread, path=["response"])
159
160 @parametrize
161 def test_raw_response_create_and_run(self, client: OpenAI) -> None:
162 response = client.beta.threads.with_raw_response.create_and_run(
163 assistant_id="string",
164 )
165 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
166 thread = response.parse()
167 assert_matches_type(Run, thread, path=["response"])
168
169
170class TestAsyncThreads:
171 strict_client = AsyncOpenAI(base_url=base_url, api_key=api_key, _strict_response_validation=True)
172 loose_client = AsyncOpenAI(base_url=base_url, api_key=api_key, _strict_response_validation=False)
173 parametrize = pytest.mark.parametrize("client", [strict_client, loose_client], ids=["strict", "loose"])
174
175 @parametrize
176 async def test_method_create(self, client: AsyncOpenAI) -> None:
177 thread = await client.beta.threads.create()
178 assert_matches_type(Thread, thread, path=["response"])
179
180 @parametrize
181 async def test_method_create_with_all_params(self, client: AsyncOpenAI) -> None:
182 thread = await client.beta.threads.create(
183 messages=[
184 {
185 "role": "user",
186 "content": "x",
187 "file_ids": ["string"],
188 "metadata": {},
189 },
190 {
191 "role": "user",
192 "content": "x",
193 "file_ids": ["string"],
194 "metadata": {},
195 },
196 {
197 "role": "user",
198 "content": "x",
199 "file_ids": ["string"],
200 "metadata": {},
201 },
202 ],
203 metadata={},
204 )
205 assert_matches_type(Thread, thread, path=["response"])
206
207 @parametrize
208 async def test_raw_response_create(self, client: AsyncOpenAI) -> None:
209 response = await client.beta.threads.with_raw_response.create()
210 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
211 thread = response.parse()
212 assert_matches_type(Thread, thread, path=["response"])
213
214 @parametrize
215 async def test_method_retrieve(self, client: AsyncOpenAI) -> None:
216 thread = await client.beta.threads.retrieve(
217 "string",
218 )
219 assert_matches_type(Thread, thread, path=["response"])
220
221 @parametrize
222 async def test_raw_response_retrieve(self, client: AsyncOpenAI) -> None:
223 response = await client.beta.threads.with_raw_response.retrieve(
224 "string",
225 )
226 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
227 thread = response.parse()
228 assert_matches_type(Thread, thread, path=["response"])
229
230 @parametrize
231 async def test_method_update(self, client: AsyncOpenAI) -> None:
232 thread = await client.beta.threads.update(
233 "string",
234 )
235 assert_matches_type(Thread, thread, path=["response"])
236
237 @parametrize
238 async def test_method_update_with_all_params(self, client: AsyncOpenAI) -> None:
239 thread = await client.beta.threads.update(
240 "string",
241 metadata={},
242 )
243 assert_matches_type(Thread, thread, path=["response"])
244
245 @parametrize
246 async def test_raw_response_update(self, client: AsyncOpenAI) -> None:
247 response = await client.beta.threads.with_raw_response.update(
248 "string",
249 )
250 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
251 thread = response.parse()
252 assert_matches_type(Thread, thread, path=["response"])
253
254 @parametrize
255 async def test_method_delete(self, client: AsyncOpenAI) -> None:
256 thread = await client.beta.threads.delete(
257 "string",
258 )
259 assert_matches_type(ThreadDeleted, thread, path=["response"])
260
261 @parametrize
262 async def test_raw_response_delete(self, client: AsyncOpenAI) -> None:
263 response = await client.beta.threads.with_raw_response.delete(
264 "string",
265 )
266 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
267 thread = response.parse()
268 assert_matches_type(ThreadDeleted, thread, path=["response"])
269
270 @parametrize
271 async def test_method_create_and_run(self, client: AsyncOpenAI) -> None:
272 thread = await client.beta.threads.create_and_run(
273 assistant_id="string",
274 )
275 assert_matches_type(Run, thread, path=["response"])
276
277 @parametrize
278 async def test_method_create_and_run_with_all_params(self, client: AsyncOpenAI) -> None:
279 thread = await client.beta.threads.create_and_run(
280 assistant_id="string",
281 instructions="string",
282 metadata={},
283 model="string",
284 thread={
285 "messages": [
286 {
287 "role": "user",
288 "content": "x",
289 "file_ids": ["string"],
290 "metadata": {},
291 },
292 {
293 "role": "user",
294 "content": "x",
295 "file_ids": ["string"],
296 "metadata": {},
297 },
298 {
299 "role": "user",
300 "content": "x",
301 "file_ids": ["string"],
302 "metadata": {},
303 },
304 ],
305 "metadata": {},
306 },
307 tools=[{"type": "code_interpreter"}, {"type": "code_interpreter"}, {"type": "code_interpreter"}],
308 )
309 assert_matches_type(Run, thread, path=["response"])
310
311 @parametrize
312 async def test_raw_response_create_and_run(self, client: AsyncOpenAI) -> None:
313 response = await client.beta.threads.with_raw_response.create_and_run(
314 assistant_id="string",
315 )
316 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
317 thread = response.parse()
318 assert_matches_type(Run, thread, path=["response"])
319