fix(vertex): SA 认证注入代理配置,细化 token 获取异常处理

- _auth_service_account 接收 endpoint 参数,通过 _get_proxy_config 解析代理
- vertex_auth 区分 TimeoutException/RequestError/通用异常,提供可读错误信息
- 新增测试覆盖代理传递和超时场景
This commit is contained in:
fawney19
2026-03-20 00:49:44 +08:00
parent 28fa03451c
commit 772f2ea601
5 changed files with 184 additions and 4 deletions

View File

@@ -0,0 +1,43 @@
from __future__ import annotations
import json
import httpx
import pytest
from src.core.vertex_auth import VertexAuthError, VertexAuthService
class _TimeoutAsyncClient:
def __init__(self, **_: object) -> None:
pass
async def __aenter__(self) -> "_TimeoutAsyncClient":
return self
async def __aexit__(self, exc_type: object, exc: object, tb: object) -> bool:
return False
async def post(self, *args: object, **kwargs: object) -> object:
raise httpx.ReadTimeout("")
@pytest.mark.asyncio
async def test_vertex_auth_timeout_error_includes_readable_message(
monkeypatch: pytest.MonkeyPatch,
) -> None:
service = VertexAuthService(
json.dumps(
{
"client_email": "svc@example.iam.gserviceaccount.com",
"private_key": "not-used-in-test",
"project_id": "demo-project",
}
)
)
monkeypatch.setattr(service, "_create_jwt", lambda: "signed-jwt")
monkeypatch.setattr("src.core.vertex_auth.httpx.AsyncClient", _TimeoutAsyncClient)
with pytest.raises(VertexAuthError, match=r"request timed out after 30s"):
await service.get_access_token(httpx_client_kwargs={"timeout": 30})