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 import Upload
13
14base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
15
16
17class TestUploads:
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 upload = client.uploads.create(
23 bytes=0,
24 filename="filename",
25 mime_type="mime_type",
26 purpose="assistants",
27 )
28 assert_matches_type(Upload, upload, path=["response"])
29
30 @parametrize
31 def test_method_create_with_all_params(self, client: OpenAI) -> None:
32 upload = client.uploads.create(
33 bytes=0,
34 filename="filename",
35 mime_type="mime_type",
36 purpose="assistants",
37 expires_after={
38 "anchor": "created_at",
39 "seconds": 3600,
40 },
41 )
42 assert_matches_type(Upload, upload, path=["response"])
43
44 @parametrize
45 def test_raw_response_create(self, client: OpenAI) -> None:
46 response = client.uploads.with_raw_response.create(
47 bytes=0,
48 filename="filename",
49 mime_type="mime_type",
50 purpose="assistants",
51 )
52
53 assert response.is_closed is True
54 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
55 upload = response.parse()
56 assert_matches_type(Upload, upload, path=["response"])
57
58 @parametrize
59 def test_streaming_response_create(self, client: OpenAI) -> None:
60 with client.uploads.with_streaming_response.create(
61 bytes=0,
62 filename="filename",
63 mime_type="mime_type",
64 purpose="assistants",
65 ) as response:
66 assert not response.is_closed
67 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
68
69 upload = response.parse()
70 assert_matches_type(Upload, upload, path=["response"])
71
72 assert cast(Any, response.is_closed) is True
73
74 @parametrize
75 def test_method_cancel(self, client: OpenAI) -> None:
76 upload = client.uploads.cancel(
77 "upload_abc123",
78 )
79 assert_matches_type(Upload, upload, path=["response"])
80
81 @parametrize
82 def test_raw_response_cancel(self, client: OpenAI) -> None:
83 response = client.uploads.with_raw_response.cancel(
84 "upload_abc123",
85 )
86
87 assert response.is_closed is True
88 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
89 upload = response.parse()
90 assert_matches_type(Upload, upload, path=["response"])
91
92 @parametrize
93 def test_streaming_response_cancel(self, client: OpenAI) -> None:
94 with client.uploads.with_streaming_response.cancel(
95 "upload_abc123",
96 ) as response:
97 assert not response.is_closed
98 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
99
100 upload = response.parse()
101 assert_matches_type(Upload, upload, path=["response"])
102
103 assert cast(Any, response.is_closed) is True
104
105 @parametrize
106 def test_path_params_cancel(self, client: OpenAI) -> None:
107 with pytest.raises(ValueError, match=r"Expected a non-empty value for `upload_id` but received ''"):
108 client.uploads.with_raw_response.cancel(
109 "",
110 )
111
112 @parametrize
113 def test_method_complete(self, client: OpenAI) -> None:
114 upload = client.uploads.complete(
115 upload_id="upload_abc123",
116 part_ids=["string"],
117 )
118 assert_matches_type(Upload, upload, path=["response"])
119
120 @parametrize
121 def test_method_complete_with_all_params(self, client: OpenAI) -> None:
122 upload = client.uploads.complete(
123 upload_id="upload_abc123",
124 part_ids=["string"],
125 md5="md5",
126 )
127 assert_matches_type(Upload, upload, path=["response"])
128
129 @parametrize
130 def test_raw_response_complete(self, client: OpenAI) -> None:
131 response = client.uploads.with_raw_response.complete(
132 upload_id="upload_abc123",
133 part_ids=["string"],
134 )
135
136 assert response.is_closed is True
137 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
138 upload = response.parse()
139 assert_matches_type(Upload, upload, path=["response"])
140
141 @parametrize
142 def test_streaming_response_complete(self, client: OpenAI) -> None:
143 with client.uploads.with_streaming_response.complete(
144 upload_id="upload_abc123",
145 part_ids=["string"],
146 ) as response:
147 assert not response.is_closed
148 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
149
150 upload = response.parse()
151 assert_matches_type(Upload, upload, path=["response"])
152
153 assert cast(Any, response.is_closed) is True
154
155 @parametrize
156 def test_path_params_complete(self, client: OpenAI) -> None:
157 with pytest.raises(ValueError, match=r"Expected a non-empty value for `upload_id` but received ''"):
158 client.uploads.with_raw_response.complete(
159 upload_id="",
160 part_ids=["string"],
161 )
162
163
164class TestAsyncUploads:
165 parametrize = pytest.mark.parametrize(
166 "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
167 )
168
169 @parametrize
170 async def test_method_create(self, async_client: AsyncOpenAI) -> None:
171 upload = await async_client.uploads.create(
172 bytes=0,
173 filename="filename",
174 mime_type="mime_type",
175 purpose="assistants",
176 )
177 assert_matches_type(Upload, upload, path=["response"])
178
179 @parametrize
180 async def test_method_create_with_all_params(self, async_client: AsyncOpenAI) -> None:
181 upload = await async_client.uploads.create(
182 bytes=0,
183 filename="filename",
184 mime_type="mime_type",
185 purpose="assistants",
186 expires_after={
187 "anchor": "created_at",
188 "seconds": 3600,
189 },
190 )
191 assert_matches_type(Upload, upload, path=["response"])
192
193 @parametrize
194 async def test_raw_response_create(self, async_client: AsyncOpenAI) -> None:
195 response = await async_client.uploads.with_raw_response.create(
196 bytes=0,
197 filename="filename",
198 mime_type="mime_type",
199 purpose="assistants",
200 )
201
202 assert response.is_closed is True
203 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
204 upload = response.parse()
205 assert_matches_type(Upload, upload, path=["response"])
206
207 @parametrize
208 async def test_streaming_response_create(self, async_client: AsyncOpenAI) -> None:
209 async with async_client.uploads.with_streaming_response.create(
210 bytes=0,
211 filename="filename",
212 mime_type="mime_type",
213 purpose="assistants",
214 ) as response:
215 assert not response.is_closed
216 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
217
218 upload = await response.parse()
219 assert_matches_type(Upload, upload, path=["response"])
220
221 assert cast(Any, response.is_closed) is True
222
223 @parametrize
224 async def test_method_cancel(self, async_client: AsyncOpenAI) -> None:
225 upload = await async_client.uploads.cancel(
226 "upload_abc123",
227 )
228 assert_matches_type(Upload, upload, path=["response"])
229
230 @parametrize
231 async def test_raw_response_cancel(self, async_client: AsyncOpenAI) -> None:
232 response = await async_client.uploads.with_raw_response.cancel(
233 "upload_abc123",
234 )
235
236 assert response.is_closed is True
237 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
238 upload = response.parse()
239 assert_matches_type(Upload, upload, path=["response"])
240
241 @parametrize
242 async def test_streaming_response_cancel(self, async_client: AsyncOpenAI) -> None:
243 async with async_client.uploads.with_streaming_response.cancel(
244 "upload_abc123",
245 ) as response:
246 assert not response.is_closed
247 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
248
249 upload = await response.parse()
250 assert_matches_type(Upload, upload, path=["response"])
251
252 assert cast(Any, response.is_closed) is True
253
254 @parametrize
255 async def test_path_params_cancel(self, async_client: AsyncOpenAI) -> None:
256 with pytest.raises(ValueError, match=r"Expected a non-empty value for `upload_id` but received ''"):
257 await async_client.uploads.with_raw_response.cancel(
258 "",
259 )
260
261 @parametrize
262 async def test_method_complete(self, async_client: AsyncOpenAI) -> None:
263 upload = await async_client.uploads.complete(
264 upload_id="upload_abc123",
265 part_ids=["string"],
266 )
267 assert_matches_type(Upload, upload, path=["response"])
268
269 @parametrize
270 async def test_method_complete_with_all_params(self, async_client: AsyncOpenAI) -> None:
271 upload = await async_client.uploads.complete(
272 upload_id="upload_abc123",
273 part_ids=["string"],
274 md5="md5",
275 )
276 assert_matches_type(Upload, upload, path=["response"])
277
278 @parametrize
279 async def test_raw_response_complete(self, async_client: AsyncOpenAI) -> None:
280 response = await async_client.uploads.with_raw_response.complete(
281 upload_id="upload_abc123",
282 part_ids=["string"],
283 )
284
285 assert response.is_closed is True
286 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
287 upload = response.parse()
288 assert_matches_type(Upload, upload, path=["response"])
289
290 @parametrize
291 async def test_streaming_response_complete(self, async_client: AsyncOpenAI) -> None:
292 async with async_client.uploads.with_streaming_response.complete(
293 upload_id="upload_abc123",
294 part_ids=["string"],
295 ) as response:
296 assert not response.is_closed
297 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
298
299 upload = await response.parse()
300 assert_matches_type(Upload, upload, path=["response"])
301
302 assert cast(Any, response.is_closed) is True
303
304 @parametrize
305 async def test_path_params_complete(self, async_client: AsyncOpenAI) -> None:
306 with pytest.raises(ValueError, match=r"Expected a non-empty value for `upload_id` but received ''"):
307 await async_client.uploads.with_raw_response.complete(
308 upload_id="",
309 part_ids=["string"],
310 )