main
1# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
2
3from __future__ import annotations
4
5import os
6from unittest import mock
7
8import pytest
9
10import openai
11from openai._exceptions import InvalidWebhookSignatureError
12
13base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
14
15# Standardized test constants (matches TypeScript implementation)
16TEST_SECRET = "whsec_RdvaYFYUXuIFuEbvZHwMfYFhUf7aMYjYcmM24+Aj40c="
17TEST_PAYLOAD = '{"id": "evt_685c059ae3a481909bdc86819b066fb6", "object": "event", "created_at": 1750861210, "type": "response.completed", "data": {"id": "resp_123"}}'
18TEST_TIMESTAMP = 1750861210 # Fixed timestamp that matches our test signature
19TEST_WEBHOOK_ID = "wh_685c059ae39c8190af8c71ed1022a24d"
20TEST_SIGNATURE = "v1,gUAg4R2hWouRZqRQG4uJypNS8YK885G838+EHb4nKBY="
21
22
23def create_test_headers(
24 timestamp: int | None = None, signature: str | None = None, webhook_id: str | None = None
25) -> dict[str, str]:
26 """Helper function to create test headers"""
27 return {
28 "webhook-signature": signature or TEST_SIGNATURE,
29 "webhook-timestamp": str(timestamp or TEST_TIMESTAMP),
30 "webhook-id": webhook_id or TEST_WEBHOOK_ID,
31 }
32
33
34class TestWebhooks:
35 parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
36
37 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
38 @parametrize
39 def test_unwrap_with_secret(self, client: openai.OpenAI) -> None:
40 headers = create_test_headers()
41 unwrapped = client.webhooks.unwrap(TEST_PAYLOAD, headers, secret=TEST_SECRET)
42 assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6"
43 assert unwrapped.created_at == 1750861210
44
45 @parametrize
46 def test_unwrap_without_secret(self, client: openai.OpenAI) -> None:
47 headers = create_test_headers()
48 with pytest.raises(ValueError, match="The webhook secret must either be set"):
49 client.webhooks.unwrap(TEST_PAYLOAD, headers)
50
51 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
52 @parametrize
53 def test_verify_signature_valid(self, client: openai.OpenAI) -> None:
54 headers = create_test_headers()
55 # Should not raise - this is a truly valid signature for this timestamp
56 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
57
58 @parametrize
59 def test_verify_signature_invalid_secret_format(self, client: openai.OpenAI) -> None:
60 headers = create_test_headers()
61 with pytest.raises(ValueError, match="The webhook secret must either be set"):
62 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=None)
63
64 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
65 @parametrize
66 def test_verify_signature_invalid(self, client: openai.OpenAI) -> None:
67 headers = create_test_headers()
68 with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"):
69 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret="invalid_secret")
70
71 @parametrize
72 def test_verify_signature_missing_webhook_signature_header(self, client: openai.OpenAI) -> None:
73 headers = create_test_headers(signature=None)
74 del headers["webhook-signature"]
75 with pytest.raises(ValueError, match="Could not find webhook-signature header"):
76 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
77
78 @parametrize
79 def test_verify_signature_missing_webhook_timestamp_header(self, client: openai.OpenAI) -> None:
80 headers = create_test_headers()
81 del headers["webhook-timestamp"]
82 with pytest.raises(ValueError, match="Could not find webhook-timestamp header"):
83 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
84
85 @parametrize
86 def test_verify_signature_missing_webhook_id_header(self, client: openai.OpenAI) -> None:
87 headers = create_test_headers()
88 del headers["webhook-id"]
89 with pytest.raises(ValueError, match="Could not find webhook-id header"):
90 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
91
92 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
93 @parametrize
94 def test_verify_signature_payload_bytes(self, client: openai.OpenAI) -> None:
95 headers = create_test_headers()
96 client.webhooks.verify_signature(TEST_PAYLOAD.encode("utf-8"), headers, secret=TEST_SECRET)
97
98 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
99 def test_unwrap_with_client_secret(self) -> None:
100 test_client = openai.OpenAI(base_url=base_url, api_key="test-api-key", webhook_secret=TEST_SECRET)
101 headers = create_test_headers()
102
103 unwrapped = test_client.webhooks.unwrap(TEST_PAYLOAD, headers)
104 assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6"
105 assert unwrapped.created_at == 1750861210
106
107 @parametrize
108 def test_verify_signature_timestamp_too_old(self, client: openai.OpenAI) -> None:
109 # Use a timestamp that's older than 5 minutes from our test timestamp
110 old_timestamp = TEST_TIMESTAMP - 400 # 6 minutes 40 seconds ago
111 headers = create_test_headers(timestamp=old_timestamp, signature="v1,dummy_signature")
112
113 with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too old"):
114 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
115
116 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
117 @parametrize
118 def test_verify_signature_timestamp_too_new(self, client: openai.OpenAI) -> None:
119 # Use a timestamp that's in the future beyond tolerance from our test timestamp
120 future_timestamp = TEST_TIMESTAMP + 400 # 6 minutes 40 seconds in the future
121 headers = create_test_headers(timestamp=future_timestamp, signature="v1,dummy_signature")
122
123 with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too new"):
124 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
125
126 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
127 @parametrize
128 def test_verify_signature_custom_tolerance(self, client: openai.OpenAI) -> None:
129 # Use a timestamp that's older than default tolerance but within custom tolerance
130 old_timestamp = TEST_TIMESTAMP - 400 # 6 minutes 40 seconds ago from test timestamp
131 headers = create_test_headers(timestamp=old_timestamp, signature="v1,dummy_signature")
132
133 # Should fail with default tolerance
134 with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too old"):
135 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
136
137 # Should also fail with custom tolerance of 10 minutes (signature won't match)
138 with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"):
139 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET, tolerance=600)
140
141 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
142 @parametrize
143 def test_verify_signature_recent_timestamp_succeeds(self, client: openai.OpenAI) -> None:
144 # Use a recent timestamp with dummy signature
145 headers = create_test_headers(signature="v1,dummy_signature")
146
147 # Should fail on signature verification (not timestamp validation)
148 with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"):
149 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
150
151 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
152 @parametrize
153 def test_verify_signature_multiple_signatures_one_valid(self, client: openai.OpenAI) -> None:
154 # Test multiple signatures: one invalid, one valid
155 multiple_signatures = f"v1,invalid_signature {TEST_SIGNATURE}"
156 headers = create_test_headers(signature=multiple_signatures)
157
158 # Should not raise when at least one signature is valid
159 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
160
161 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
162 @parametrize
163 def test_verify_signature_multiple_signatures_all_invalid(self, client: openai.OpenAI) -> None:
164 # Test multiple invalid signatures
165 multiple_invalid_signatures = "v1,invalid_signature1 v1,invalid_signature2"
166 headers = create_test_headers(signature=multiple_invalid_signatures)
167
168 with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"):
169 client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
170
171
172class TestAsyncWebhooks:
173 parametrize = pytest.mark.parametrize(
174 "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
175 )
176
177 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
178 @parametrize
179 async def test_unwrap_with_secret(self, async_client: openai.AsyncOpenAI) -> None:
180 headers = create_test_headers()
181 unwrapped = async_client.webhooks.unwrap(TEST_PAYLOAD, headers, secret=TEST_SECRET)
182 assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6"
183 assert unwrapped.created_at == 1750861210
184
185 @parametrize
186 async def test_unwrap_without_secret(self, async_client: openai.AsyncOpenAI) -> None:
187 headers = create_test_headers()
188 with pytest.raises(ValueError, match="The webhook secret must either be set"):
189 async_client.webhooks.unwrap(TEST_PAYLOAD, headers)
190
191 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
192 @parametrize
193 async def test_verify_signature_valid(self, async_client: openai.AsyncOpenAI) -> None:
194 headers = create_test_headers()
195 # Should not raise - this is a truly valid signature for this timestamp
196 async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
197
198 @parametrize
199 async def test_verify_signature_invalid_secret_format(self, async_client: openai.AsyncOpenAI) -> None:
200 headers = create_test_headers()
201 with pytest.raises(ValueError, match="The webhook secret must either be set"):
202 async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=None)
203
204 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
205 @parametrize
206 async def test_verify_signature_invalid(self, async_client: openai.AsyncOpenAI) -> None:
207 headers = create_test_headers()
208 with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"):
209 async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret="invalid_secret")
210
211 @parametrize
212 async def test_verify_signature_missing_webhook_signature_header(self, async_client: openai.AsyncOpenAI) -> None:
213 headers = create_test_headers()
214 del headers["webhook-signature"]
215 with pytest.raises(ValueError, match="Could not find webhook-signature header"):
216 async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
217
218 @parametrize
219 async def test_verify_signature_missing_webhook_timestamp_header(self, async_client: openai.AsyncOpenAI) -> None:
220 headers = create_test_headers()
221 del headers["webhook-timestamp"]
222 with pytest.raises(ValueError, match="Could not find webhook-timestamp header"):
223 async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
224
225 @parametrize
226 async def test_verify_signature_missing_webhook_id_header(self, async_client: openai.AsyncOpenAI) -> None:
227 headers = create_test_headers()
228 del headers["webhook-id"]
229 with pytest.raises(ValueError, match="Could not find webhook-id header"):
230 async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
231
232 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
233 @parametrize
234 async def test_verify_signature_payload_bytes(self, async_client: openai.AsyncOpenAI) -> None:
235 headers = create_test_headers()
236 async_client.webhooks.verify_signature(TEST_PAYLOAD.encode("utf-8"), headers, secret=TEST_SECRET)
237
238 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
239 async def test_unwrap_with_client_secret(self) -> None:
240 test_async_client = openai.AsyncOpenAI(base_url=base_url, api_key="test-api-key", webhook_secret=TEST_SECRET)
241 headers = create_test_headers()
242
243 unwrapped = test_async_client.webhooks.unwrap(TEST_PAYLOAD, headers)
244 assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6"
245 assert unwrapped.created_at == 1750861210
246
247 @parametrize
248 async def test_verify_signature_timestamp_too_old(self, async_client: openai.AsyncOpenAI) -> None:
249 # Use a timestamp that's older than 5 minutes from our test timestamp
250 old_timestamp = TEST_TIMESTAMP - 400 # 6 minutes 40 seconds ago
251 headers = create_test_headers(timestamp=old_timestamp, signature="v1,dummy_signature")
252
253 with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too old"):
254 async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
255
256 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
257 @parametrize
258 async def test_verify_signature_timestamp_too_new(self, async_client: openai.AsyncOpenAI) -> None:
259 # Use a timestamp that's in the future beyond tolerance from our test timestamp
260 future_timestamp = TEST_TIMESTAMP + 400 # 6 minutes 40 seconds in the future
261 headers = create_test_headers(timestamp=future_timestamp, signature="v1,dummy_signature")
262
263 with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too new"):
264 async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
265
266 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
267 @parametrize
268 async def test_verify_signature_multiple_signatures_one_valid(self, async_client: openai.AsyncOpenAI) -> None:
269 # Test multiple signatures: one invalid, one valid
270 multiple_signatures = f"v1,invalid_signature {TEST_SIGNATURE}"
271 headers = create_test_headers(signature=multiple_signatures)
272
273 # Should not raise when at least one signature is valid
274 async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)
275
276 @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP))
277 @parametrize
278 async def test_verify_signature_multiple_signatures_all_invalid(self, async_client: openai.AsyncOpenAI) -> None:
279 # Test multiple invalid signatures
280 multiple_invalid_signatures = "v1,invalid_signature1 v1,invalid_signature2"
281 headers = create_test_headers(signature=multiple_invalid_signatures)
282
283 with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"):
284 async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)