(respx_mock: MockRouter)
| 323 | |
| 324 | @pytest.mark.respx() |
| 325 | def test_token_provider_refresh_sync(respx_mock: MockRouter) -> None: |
| 326 | respx_mock.post("https://example.com/openai/v1/responses").mock( |
| 327 | side_effect=[ |
| 328 | httpx.Response(500, json={"error": "server error"}), |
| 329 | httpx.Response(200, json=RESPONSE_BODY), |
| 330 | ] |
| 331 | ) |
| 332 | tokens = iter(["first", "second"]) |
| 333 | client = BedrockOpenAI( |
| 334 | base_url="https://example.com/openai/v1", |
| 335 | bedrock_token_provider=lambda: next(tokens), |
| 336 | http_client=httpx.Client(trust_env=False), |
| 337 | max_retries=1, |
| 338 | ) |
| 339 | |
| 340 | client.responses.create(model="gpt-4o", input="hello") |
| 341 | |
| 342 | calls = cast("list[MockRequestCall]", respx_mock.calls) |
| 343 | assert calls[0].request.headers["Authorization"] == "Bearer first" |
| 344 | assert calls[1].request.headers["Authorization"] == "Bearer second" |
| 345 | |
| 346 | |
| 347 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected