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 (
13 ContainerListResponse,
14 ContainerCreateResponse,
15 ContainerRetrieveResponse,
16)
17from openai.pagination import SyncCursorPage, AsyncCursorPage
18
19base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
20
21
22class TestContainers:
23 parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
24
25 @parametrize
26 def test_method_create(self, client: OpenAI) -> None:
27 container = client.containers.create(
28 name="name",
29 )
30 assert_matches_type(ContainerCreateResponse, container, path=["response"])
31
32 @parametrize
33 def test_method_create_with_all_params(self, client: OpenAI) -> None:
34 container = client.containers.create(
35 name="name",
36 expires_after={
37 "anchor": "last_active_at",
38 "minutes": 0,
39 },
40 file_ids=["string"],
41 memory_limit="1g",
42 )
43 assert_matches_type(ContainerCreateResponse, container, path=["response"])
44
45 @parametrize
46 def test_raw_response_create(self, client: OpenAI) -> None:
47 response = client.containers.with_raw_response.create(
48 name="name",
49 )
50
51 assert response.is_closed is True
52 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
53 container = response.parse()
54 assert_matches_type(ContainerCreateResponse, container, path=["response"])
55
56 @parametrize
57 def test_streaming_response_create(self, client: OpenAI) -> None:
58 with client.containers.with_streaming_response.create(
59 name="name",
60 ) as response:
61 assert not response.is_closed
62 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
63
64 container = response.parse()
65 assert_matches_type(ContainerCreateResponse, container, path=["response"])
66
67 assert cast(Any, response.is_closed) is True
68
69 @parametrize
70 def test_method_retrieve(self, client: OpenAI) -> None:
71 container = client.containers.retrieve(
72 "container_id",
73 )
74 assert_matches_type(ContainerRetrieveResponse, container, path=["response"])
75
76 @parametrize
77 def test_raw_response_retrieve(self, client: OpenAI) -> None:
78 response = client.containers.with_raw_response.retrieve(
79 "container_id",
80 )
81
82 assert response.is_closed is True
83 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
84 container = response.parse()
85 assert_matches_type(ContainerRetrieveResponse, container, path=["response"])
86
87 @parametrize
88 def test_streaming_response_retrieve(self, client: OpenAI) -> None:
89 with client.containers.with_streaming_response.retrieve(
90 "container_id",
91 ) as response:
92 assert not response.is_closed
93 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
94
95 container = response.parse()
96 assert_matches_type(ContainerRetrieveResponse, container, path=["response"])
97
98 assert cast(Any, response.is_closed) is True
99
100 @parametrize
101 def test_path_params_retrieve(self, client: OpenAI) -> None:
102 with pytest.raises(ValueError, match=r"Expected a non-empty value for `container_id` but received ''"):
103 client.containers.with_raw_response.retrieve(
104 "",
105 )
106
107 @parametrize
108 def test_method_list(self, client: OpenAI) -> None:
109 container = client.containers.list()
110 assert_matches_type(SyncCursorPage[ContainerListResponse], container, path=["response"])
111
112 @parametrize
113 def test_method_list_with_all_params(self, client: OpenAI) -> None:
114 container = client.containers.list(
115 after="after",
116 limit=0,
117 order="asc",
118 )
119 assert_matches_type(SyncCursorPage[ContainerListResponse], container, path=["response"])
120
121 @parametrize
122 def test_raw_response_list(self, client: OpenAI) -> None:
123 response = client.containers.with_raw_response.list()
124
125 assert response.is_closed is True
126 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
127 container = response.parse()
128 assert_matches_type(SyncCursorPage[ContainerListResponse], container, path=["response"])
129
130 @parametrize
131 def test_streaming_response_list(self, client: OpenAI) -> None:
132 with client.containers.with_streaming_response.list() as response:
133 assert not response.is_closed
134 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
135
136 container = response.parse()
137 assert_matches_type(SyncCursorPage[ContainerListResponse], container, path=["response"])
138
139 assert cast(Any, response.is_closed) is True
140
141 @parametrize
142 def test_method_delete(self, client: OpenAI) -> None:
143 container = client.containers.delete(
144 "container_id",
145 )
146 assert container is None
147
148 @parametrize
149 def test_raw_response_delete(self, client: OpenAI) -> None:
150 response = client.containers.with_raw_response.delete(
151 "container_id",
152 )
153
154 assert response.is_closed is True
155 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
156 container = response.parse()
157 assert container is None
158
159 @parametrize
160 def test_streaming_response_delete(self, client: OpenAI) -> None:
161 with client.containers.with_streaming_response.delete(
162 "container_id",
163 ) as response:
164 assert not response.is_closed
165 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
166
167 container = response.parse()
168 assert container is None
169
170 assert cast(Any, response.is_closed) is True
171
172 @parametrize
173 def test_path_params_delete(self, client: OpenAI) -> None:
174 with pytest.raises(ValueError, match=r"Expected a non-empty value for `container_id` but received ''"):
175 client.containers.with_raw_response.delete(
176 "",
177 )
178
179
180class TestAsyncContainers:
181 parametrize = pytest.mark.parametrize(
182 "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
183 )
184
185 @parametrize
186 async def test_method_create(self, async_client: AsyncOpenAI) -> None:
187 container = await async_client.containers.create(
188 name="name",
189 )
190 assert_matches_type(ContainerCreateResponse, container, path=["response"])
191
192 @parametrize
193 async def test_method_create_with_all_params(self, async_client: AsyncOpenAI) -> None:
194 container = await async_client.containers.create(
195 name="name",
196 expires_after={
197 "anchor": "last_active_at",
198 "minutes": 0,
199 },
200 file_ids=["string"],
201 memory_limit="1g",
202 )
203 assert_matches_type(ContainerCreateResponse, container, path=["response"])
204
205 @parametrize
206 async def test_raw_response_create(self, async_client: AsyncOpenAI) -> None:
207 response = await async_client.containers.with_raw_response.create(
208 name="name",
209 )
210
211 assert response.is_closed is True
212 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
213 container = response.parse()
214 assert_matches_type(ContainerCreateResponse, container, path=["response"])
215
216 @parametrize
217 async def test_streaming_response_create(self, async_client: AsyncOpenAI) -> None:
218 async with async_client.containers.with_streaming_response.create(
219 name="name",
220 ) as response:
221 assert not response.is_closed
222 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
223
224 container = await response.parse()
225 assert_matches_type(ContainerCreateResponse, container, path=["response"])
226
227 assert cast(Any, response.is_closed) is True
228
229 @parametrize
230 async def test_method_retrieve(self, async_client: AsyncOpenAI) -> None:
231 container = await async_client.containers.retrieve(
232 "container_id",
233 )
234 assert_matches_type(ContainerRetrieveResponse, container, path=["response"])
235
236 @parametrize
237 async def test_raw_response_retrieve(self, async_client: AsyncOpenAI) -> None:
238 response = await async_client.containers.with_raw_response.retrieve(
239 "container_id",
240 )
241
242 assert response.is_closed is True
243 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
244 container = response.parse()
245 assert_matches_type(ContainerRetrieveResponse, container, path=["response"])
246
247 @parametrize
248 async def test_streaming_response_retrieve(self, async_client: AsyncOpenAI) -> None:
249 async with async_client.containers.with_streaming_response.retrieve(
250 "container_id",
251 ) as response:
252 assert not response.is_closed
253 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
254
255 container = await response.parse()
256 assert_matches_type(ContainerRetrieveResponse, container, path=["response"])
257
258 assert cast(Any, response.is_closed) is True
259
260 @parametrize
261 async def test_path_params_retrieve(self, async_client: AsyncOpenAI) -> None:
262 with pytest.raises(ValueError, match=r"Expected a non-empty value for `container_id` but received ''"):
263 await async_client.containers.with_raw_response.retrieve(
264 "",
265 )
266
267 @parametrize
268 async def test_method_list(self, async_client: AsyncOpenAI) -> None:
269 container = await async_client.containers.list()
270 assert_matches_type(AsyncCursorPage[ContainerListResponse], container, path=["response"])
271
272 @parametrize
273 async def test_method_list_with_all_params(self, async_client: AsyncOpenAI) -> None:
274 container = await async_client.containers.list(
275 after="after",
276 limit=0,
277 order="asc",
278 )
279 assert_matches_type(AsyncCursorPage[ContainerListResponse], container, path=["response"])
280
281 @parametrize
282 async def test_raw_response_list(self, async_client: AsyncOpenAI) -> None:
283 response = await async_client.containers.with_raw_response.list()
284
285 assert response.is_closed is True
286 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
287 container = response.parse()
288 assert_matches_type(AsyncCursorPage[ContainerListResponse], container, path=["response"])
289
290 @parametrize
291 async def test_streaming_response_list(self, async_client: AsyncOpenAI) -> None:
292 async with async_client.containers.with_streaming_response.list() as response:
293 assert not response.is_closed
294 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
295
296 container = await response.parse()
297 assert_matches_type(AsyncCursorPage[ContainerListResponse], container, path=["response"])
298
299 assert cast(Any, response.is_closed) is True
300
301 @parametrize
302 async def test_method_delete(self, async_client: AsyncOpenAI) -> None:
303 container = await async_client.containers.delete(
304 "container_id",
305 )
306 assert container is None
307
308 @parametrize
309 async def test_raw_response_delete(self, async_client: AsyncOpenAI) -> None:
310 response = await async_client.containers.with_raw_response.delete(
311 "container_id",
312 )
313
314 assert response.is_closed is True
315 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
316 container = response.parse()
317 assert container is None
318
319 @parametrize
320 async def test_streaming_response_delete(self, async_client: AsyncOpenAI) -> None:
321 async with async_client.containers.with_streaming_response.delete(
322 "container_id",
323 ) as response:
324 assert not response.is_closed
325 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
326
327 container = await response.parse()
328 assert container is None
329
330 assert cast(Any, response.is_closed) is True
331
332 @parametrize
333 async def test_path_params_delete(self, async_client: AsyncOpenAI) -> None:
334 with pytest.raises(ValueError, match=r"Expected a non-empty value for `container_id` but received ''"):
335 await async_client.containers.with_raw_response.delete(
336 "",
337 )