b523b4fd21
- 薄自建 LLM 网关：OpenAI 兼容适配器(DeepSeek) + instructor 结构化输出 + usage_ledger 记账 + 档位路由 - 记忆服务 assemble：确定性选择(显式+主角+近况) + 渲染卡 + 缓存断点(中性文本) - LangGraph 写章节点 + Postgres checkpointer + SSE 归一(token/done/error) - API：立项 + 写章 draft(SSE) + PUT 自动保存 + 提供商凭据(Fernet 加密/测试连接) - 前端：AppShell + 作品库 + 5 步立项向导 + 写作工作台(流式打字机+自动保存) + 设置页 - M1 E2E：真实 DB + mock 网关零 token 走通闭环
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""T1.7 端点测试：凭据 upsert/列表/探测（内存替身，无 DB/无网络）。
|
||
|
||
DoD：掩码往返、test-connection 经可注入探测回能力矩阵、明文绝不出响应。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from fakes_providers import FakeCredentialStore, FakeProviderProbe
|
||
from fastapi.testclient import TestClient
|
||
from ww_api.security.credentials import decrypt_api_key
|
||
from ww_api.services.credentials import STUB_OWNER_ID
|
||
|
||
|
||
def test_get_empty_providers(client: TestClient) -> None:
|
||
resp = client.get("/settings/providers")
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body == {"providers": [], "tier_routing": []}
|
||
|
||
|
||
def test_put_credential_encrypts_and_masks(
|
||
client: TestClient, store: FakeCredentialStore, enc_key: str
|
||
) -> None:
|
||
# Act
|
||
resp = client.put(
|
||
"/settings/providers",
|
||
json={"credentials": [{"provider": "deepseek", "api_key": "sk-secret-9999"}]},
|
||
)
|
||
# Assert: 响应含掩码，绝不含明文
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert "sk-secret-9999" not in resp.text
|
||
assert body["providers"][0]["provider"] == "deepseek"
|
||
# 存的是密文且能解回明文（往返）
|
||
blob = store.creds[(STUB_OWNER_ID, "deepseek")]
|
||
assert b"sk-secret-9999" not in blob
|
||
assert decrypt_api_key(blob, key=enc_key) == "sk-secret-9999"
|
||
|
||
|
||
def test_put_is_idempotent(client: TestClient, store: FakeCredentialStore) -> None:
|
||
payload = {"credentials": [{"provider": "deepseek", "api_key": "sk-a"}]}
|
||
client.put("/settings/providers", json=payload)
|
||
client.put(
|
||
"/settings/providers", json={"credentials": [{"provider": "deepseek", "api_key": "sk-b"}]}
|
||
)
|
||
# 同 (owner, provider) 只一条
|
||
assert len([k for k in store.creds if k[1] == "deepseek"]) == 1
|
||
|
||
|
||
def test_put_tier_routing(client: TestClient, store: FakeCredentialStore) -> None:
|
||
resp = client.put(
|
||
"/settings/providers",
|
||
json={
|
||
"tier_routing": [
|
||
{
|
||
"tier": "writer",
|
||
"provider": "deepseek",
|
||
"model": "deepseek-chat",
|
||
"fallback": ["kimi"],
|
||
}
|
||
]
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
routing = resp.json()["tier_routing"]
|
||
assert routing[0] == {
|
||
"tier": "writer",
|
||
"provider": "deepseek",
|
||
"model": "deepseek-chat",
|
||
"fallback": ["kimi"],
|
||
}
|
||
|
||
|
||
def test_test_connection_returns_capabilities(client: TestClient, probe: FakeProviderProbe) -> None:
|
||
resp = client.post("/settings/providers/test", json={"provider": "deepseek"})
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["provider"] == "deepseek"
|
||
assert body["ok"] is True
|
||
assert body["capabilities"] == {
|
||
"structured_output": True,
|
||
"prefix_cache": True,
|
||
"thinking": False,
|
||
}
|
||
assert probe.calls == [(STUB_OWNER_ID, "deepseek")]
|
||
|
||
|
||
def test_put_validation_rejects_blank_provider(client: TestClient) -> None:
|
||
resp = client.put(
|
||
"/settings/providers",
|
||
json={"credentials": [{"provider": "", "api_key": "sk-x"}]},
|
||
)
|
||
assert resp.status_code == 422
|