`api_key=volcengine.api_key` 这类**依赖注入传参**被 KEY-ASSIGN 判为疑似凭证。 这是规则的真缺口而非代码问题:行里没有任何秘密,而做 DI 的代码到处都是这种 写法——绕过一次就得在每个调用点绕过,那等于把规则废掉。故改规则。 新增 ALLOW_REFERENCE,刻意收得很紧以免开后门: - 值必须是**未加引号**的小写起首标识符引用(真密钥要么带引号,要么不是标识符形状) - 只允许**恰好一个点且随即行尾** - 连字符不在字符集内 对抗性夹具验证(三真三假): - 拦下 `api_key = "sk-T3Blbk…"`(OpenAI 形态) - 拦下 `secret_key: "AKID…"`(腾讯云形态) - 拦下 `auth_token=eyJ….eyJ….Sfl…`(**三段式 JWT——两个点,故不落进放行**) - 放行 `api_key=lowercase.reference` / `api_key = settings.api_key` / `api_token: config.authToken` 原 5 形态泄漏夹具复验仍 8 条命中,一条不少。 另更正口径:脚本头与 git-workflow.md 仍写「三仓同构」,实际已是四仓—— patbond-ai 是唯一持有 AI 服务商凭证的仓,漏同步它风险最大。四仓 md5 已一致 (4945940093b4b8413b7d3eec8bf6a5ec),四仓 --all 全绿。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,153 @@
|
||||
"""`ArkClient` 的测试——只测传输层:鉴权头、错误解包、连接生命周期。
|
||||
|
||||
**刻意不测任何视频业务**:本 client 不认识「视频」,测试也不该。视频相关的断言
|
||||
在 `test_video.py`。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from app.domain.generation import Retry, UserFacing
|
||||
from app.providers.volcengine.client import ArkClient, ArkError
|
||||
|
||||
#: 显式的假值。防泄漏扫描会拦下 `api_key = "..."` 形态(规则正确——这正是真实
|
||||
#: 密钥最常见的泄漏形态),故夹具用 ALLOW 清单认可的占位词。
|
||||
FAKE_API_KEY = "dummy-api-key-for-tests"
|
||||
|
||||
BASE_URL = "https://ark.example.invalid/api/v3"
|
||||
|
||||
|
||||
def client_with(handler) -> ArkClient:
|
||||
return ArkClient(
|
||||
api_key=FAKE_API_KEY,
|
||||
base_url=BASE_URL,
|
||||
client=httpx.Client(transport=httpx.MockTransport(handler)),
|
||||
)
|
||||
|
||||
|
||||
class TestConstruction:
|
||||
def test_missing_api_key_fails_at_construction_not_at_first_call(self):
|
||||
# 缺密钥是配置错误。让它表现为一串莫名的 401 会浪费排查时间。
|
||||
with pytest.raises(ValueError, match="ARK_API_KEY"):
|
||||
ArkClient(api_key="", base_url=BASE_URL)
|
||||
|
||||
def test_trailing_slash_in_base_url_does_not_double_up(self):
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
assert "//contents" not in str(request.url)
|
||||
return httpx.Response(200, json={})
|
||||
|
||||
ArkClient(
|
||||
api_key=FAKE_API_KEY,
|
||||
base_url=BASE_URL + "/",
|
||||
client=httpx.Client(transport=httpx.MockTransport(handler)),
|
||||
).get("/contents/x")
|
||||
|
||||
|
||||
class TestAuthAndTransport:
|
||||
def test_bearer_token_and_content_type_are_sent(self):
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
assert request.headers["Authorization"] == f"Bearer {FAKE_API_KEY}"
|
||||
assert request.headers["Content-Type"] == "application/json"
|
||||
return httpx.Response(200, json={"ok": True})
|
||||
|
||||
assert client_with(handler).post("/x", {}) == {"ok": True}
|
||||
|
||||
def test_get_and_post_hit_the_expected_method(self):
|
||||
seen: list[str] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
seen.append(request.method)
|
||||
return httpx.Response(200, json={})
|
||||
|
||||
client = client_with(handler)
|
||||
client.post("/x", {})
|
||||
client.get("/x")
|
||||
assert seen == ["POST", "GET"]
|
||||
|
||||
def test_network_failure_is_classified_retryable(self):
|
||||
def handler(request):
|
||||
raise httpx.ConnectError("connection refused")
|
||||
|
||||
with pytest.raises(ArkError) as exc:
|
||||
client_with(handler).get("/x")
|
||||
assert exc.value.failure.retry is Retry.BACKOFF
|
||||
assert exc.value.http_status is None
|
||||
|
||||
|
||||
class TestErrorUnwrapping:
|
||||
def test_vendor_error_body_is_classified(self):
|
||||
def handler(request):
|
||||
return httpx.Response(
|
||||
429,
|
||||
json={
|
||||
"error": {
|
||||
"code": "RateLimitExceeded.EndpointRPMExceeded",
|
||||
"message": "RPM limit exceeded",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
with pytest.raises(ArkError) as exc:
|
||||
client_with(handler).get("/x")
|
||||
assert exc.value.failure.retry is Retry.BACKOFF
|
||||
assert exc.value.failure.user_facing is UserFacing.RATE_LIMITED
|
||||
assert (
|
||||
exc.value.failure.provider_code == "RateLimitExceeded.EndpointRPMExceeded"
|
||||
)
|
||||
assert exc.value.http_status == 429
|
||||
|
||||
def test_vendor_message_is_preserved_for_troubleshooting(self):
|
||||
def handler(request):
|
||||
return httpx.Response(
|
||||
401,
|
||||
json={
|
||||
"error": {"code": "AuthenticationError", "message": "key invalid"}
|
||||
},
|
||||
)
|
||||
|
||||
with pytest.raises(ArkError) as exc:
|
||||
client_with(handler).get("/x")
|
||||
assert "key invalid" in exc.value.failure.message
|
||||
|
||||
def test_non_json_error_body_is_not_swallowed(self):
|
||||
# 打到错误地址时常见:网关返回 HTML 错误页。静默会让排查绕远路。
|
||||
def handler(request):
|
||||
return httpx.Response(502, text="<html>bad gateway</html>")
|
||||
|
||||
with pytest.raises(ArkError) as exc:
|
||||
client_with(handler).get("/x")
|
||||
assert "bad gateway" in exc.value.failure.message
|
||||
assert exc.value.failure.retry is Retry.BACKOFF
|
||||
|
||||
def test_non_json_success_body_is_an_error_too(self):
|
||||
def handler(request):
|
||||
return httpx.Response(200, text="not json")
|
||||
|
||||
with pytest.raises(ArkError, match="不是合法 JSON"):
|
||||
client_with(handler).get("/x")
|
||||
|
||||
|
||||
class TestConnectionOwnership:
|
||||
def test_injected_client_is_not_closed_by_us(self):
|
||||
"""注入进来的连接池归调用方管理。
|
||||
|
||||
在这里关掉会让共享同一个 httpx.Client 的其他适配器一起失效——接了第二个
|
||||
火山接口后这会变成一个极难查的偶发失败。
|
||||
"""
|
||||
injected = httpx.Client(
|
||||
transport=httpx.MockTransport(lambda r: httpx.Response(200, json={}))
|
||||
)
|
||||
ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL, client=injected).close()
|
||||
assert not injected.is_closed
|
||||
|
||||
def test_own_client_is_closed(self):
|
||||
client = ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL)
|
||||
client.close()
|
||||
assert client._client.is_closed
|
||||
|
||||
def test_usable_as_context_manager(self):
|
||||
with ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL) as client:
|
||||
inner = client._client
|
||||
assert inner.is_closed
|
||||
Reference in New Issue
Block a user