| 118 | |
| 119 | @respx.mock |
| 120 | def test_workload_identity_exchange_error() -> None: |
| 121 | exchange_route = respx.post("https://auth.openai.com/oauth/token").mock( |
| 122 | return_value=httpx.Response( |
| 123 | 401, |
| 124 | json={ |
| 125 | "error": "invalid_grant", |
| 126 | "error_description": "No service account mapping found for the provided service_account_id.", |
| 127 | }, |
| 128 | ) |
| 129 | ) |
| 130 | api_route = respx.get("https://api.openai.com/v1/models").mock( |
| 131 | return_value=httpx.Response(200, json={"data": [], "object": "list"}) |
| 132 | ) |
| 133 | |
| 134 | client = OpenAI( |
| 135 | workload_identity={ |
| 136 | "identity_provider_id": "idp_123", |
| 137 | "service_account_id": "sa_123", |
| 138 | "provider": { |
| 139 | "get_token": lambda: "fake_subject_token", |
| 140 | "token_type": "jwt", |
| 141 | }, |
| 142 | }, |
| 143 | ) |
| 144 | |
| 145 | with pytest.raises(OAuthError) as exc: |
| 146 | client.models.list() |
| 147 | |
| 148 | assert exc.value.message == "No service account mapping found for the provided service_account_id." |
| 149 | assert exc.value.error == "invalid_grant" |
| 150 | assert exc.value.status_code == 401 |
| 151 | assert exchange_route.call_count == 1 |
| 152 | assert api_route.call_count == 0 |
| 153 | |
| 154 | |
| 155 | def test_k8s_service_account_token_provider(tmp_path: Path) -> None: |