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
dev/codex/pin-setup-rye-workflows-main

Branches

Tags

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

Clone

HTTPS

Download ZIP

openai-python/tests/api_resources/beta/chatkit

tests/api_resources/beta/chatkit/test_sessions.py

230lines · modecode

RawDownload
Latest commit unavailable.
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.beta.chatkit import (
13 ChatSession,
14)
15
16base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
17
18
19class TestSessions:
20 parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
21
22 @parametrize
23 def test_method_create(self, client: OpenAI) -> None:
24 session = client.beta.chatkit.sessions.create(
25 user="x",
26 workflow={"id": "id"},
27 )
28 assert_matches_type(ChatSession, session, path=["response"])
29
30 @parametrize
31 def test_method_create_with_all_params(self, client: OpenAI) -> None:
32 session = client.beta.chatkit.sessions.create(
33 user="x",
34 workflow={
35 "id": "id",
36 "state_variables": {"foo": "string"},
37 "tracing": {"enabled": True},
38 "version": "version",
39 },
40 chatkit_configuration={
41 "automatic_thread_titling": {"enabled": True},
42 "file_upload": {
43 "enabled": True,
44 "max_file_size": 1,
45 "max_files": 1,
46 },
47 "history": {
48 "enabled": True,
49 "recent_threads": 1,
50 },
51 },
52 expires_after={
53 "anchor": "created_at",
54 "seconds": 1,
55 },
56 rate_limits={"max_requests_per_1_minute": 1},
57 )
58 assert_matches_type(ChatSession, session, path=["response"])
59
60 @parametrize
61 def test_raw_response_create(self, client: OpenAI) -> None:
62 response = client.beta.chatkit.sessions.with_raw_response.create(
63 user="x",
64 workflow={"id": "id"},
65 )
66
67 assert response.is_closed is True
68 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
69 session = response.parse()
70 assert_matches_type(ChatSession, session, path=["response"])
71
72 @parametrize
73 def test_streaming_response_create(self, client: OpenAI) -> None:
74 with client.beta.chatkit.sessions.with_streaming_response.create(
75 user="x",
76 workflow={"id": "id"},
77 ) as response:
78 assert not response.is_closed
79 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
80
81 session = response.parse()
82 assert_matches_type(ChatSession, session, path=["response"])
83
84 assert cast(Any, response.is_closed) is True
85
86 @parametrize
87 def test_method_cancel(self, client: OpenAI) -> None:
88 session = client.beta.chatkit.sessions.cancel(
89 "cksess_123",
90 )
91 assert_matches_type(ChatSession, session, path=["response"])
92
93 @parametrize
94 def test_raw_response_cancel(self, client: OpenAI) -> None:
95 response = client.beta.chatkit.sessions.with_raw_response.cancel(
96 "cksess_123",
97 )
98
99 assert response.is_closed is True
100 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
101 session = response.parse()
102 assert_matches_type(ChatSession, session, path=["response"])
103
104 @parametrize
105 def test_streaming_response_cancel(self, client: OpenAI) -> None:
106 with client.beta.chatkit.sessions.with_streaming_response.cancel(
107 "cksess_123",
108 ) as response:
109 assert not response.is_closed
110 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
111
112 session = response.parse()
113 assert_matches_type(ChatSession, session, path=["response"])
114
115 assert cast(Any, response.is_closed) is True
116
117 @parametrize
118 def test_path_params_cancel(self, client: OpenAI) -> None:
119 with pytest.raises(ValueError, match=r"Expected a non-empty value for `session_id` but received ''"):
120 client.beta.chatkit.sessions.with_raw_response.cancel(
121 "",
122 )
123
124
125class TestAsyncSessions:
126 parametrize = pytest.mark.parametrize(
127 "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
128 )
129
130 @parametrize
131 async def test_method_create(self, async_client: AsyncOpenAI) -> None:
132 session = await async_client.beta.chatkit.sessions.create(
133 user="x",
134 workflow={"id": "id"},
135 )
136 assert_matches_type(ChatSession, session, path=["response"])
137
138 @parametrize
139 async def test_method_create_with_all_params(self, async_client: AsyncOpenAI) -> None:
140 session = await async_client.beta.chatkit.sessions.create(
141 user="x",
142 workflow={
143 "id": "id",
144 "state_variables": {"foo": "string"},
145 "tracing": {"enabled": True},
146 "version": "version",
147 },
148 chatkit_configuration={
149 "automatic_thread_titling": {"enabled": True},
150 "file_upload": {
151 "enabled": True,
152 "max_file_size": 1,
153 "max_files": 1,
154 },
155 "history": {
156 "enabled": True,
157 "recent_threads": 1,
158 },
159 },
160 expires_after={
161 "anchor": "created_at",
162 "seconds": 1,
163 },
164 rate_limits={"max_requests_per_1_minute": 1},
165 )
166 assert_matches_type(ChatSession, session, path=["response"])
167
168 @parametrize
169 async def test_raw_response_create(self, async_client: AsyncOpenAI) -> None:
170 response = await async_client.beta.chatkit.sessions.with_raw_response.create(
171 user="x",
172 workflow={"id": "id"},
173 )
174
175 assert response.is_closed is True
176 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
177 session = response.parse()
178 assert_matches_type(ChatSession, session, path=["response"])
179
180 @parametrize
181 async def test_streaming_response_create(self, async_client: AsyncOpenAI) -> None:
182 async with async_client.beta.chatkit.sessions.with_streaming_response.create(
183 user="x",
184 workflow={"id": "id"},
185 ) as response:
186 assert not response.is_closed
187 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
188
189 session = await response.parse()
190 assert_matches_type(ChatSession, session, path=["response"])
191
192 assert cast(Any, response.is_closed) is True
193
194 @parametrize
195 async def test_method_cancel(self, async_client: AsyncOpenAI) -> None:
196 session = await async_client.beta.chatkit.sessions.cancel(
197 "cksess_123",
198 )
199 assert_matches_type(ChatSession, session, path=["response"])
200
201 @parametrize
202 async def test_raw_response_cancel(self, async_client: AsyncOpenAI) -> None:
203 response = await async_client.beta.chatkit.sessions.with_raw_response.cancel(
204 "cksess_123",
205 )
206
207 assert response.is_closed is True
208 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
209 session = response.parse()
210 assert_matches_type(ChatSession, session, path=["response"])
211
212 @parametrize
213 async def test_streaming_response_cancel(self, async_client: AsyncOpenAI) -> None:
214 async with async_client.beta.chatkit.sessions.with_streaming_response.cancel(
215 "cksess_123",
216 ) as response:
217 assert not response.is_closed
218 assert response.http_request.headers.get("X-Stainless-Lang") == "python"
219
220 session = await response.parse()
221 assert_matches_type(ChatSession, session, path=["response"])
222
223 assert cast(Any, response.is_closed) is True
224
225 @parametrize
226 async def test_path_params_cancel(self, async_client: AsyncOpenAI) -> None:
227 with pytest.raises(ValueError, match=r"Expected a non-empty value for `session_id` but received ''"):
228 await async_client.beta.chatkit.sessions.with_raw_response.cancel(
229 "",
230 )
231