345cc73965
P0-1 SqlCredentialStore/save_draft 由自提交改 flush，端点/服务统一 commit （新增 CredentialStore.commit() 统一提交点；token 刷新落库显式提交）； 补多凭据一请求中途失败整体回滚集成测试。 P0-2 启动校验 _fernet(enc_key) 快速失败 + catch-all Exception → ErrorEnvelope； credential_enc_key 改 SecretStr。 P0-3 run_job 异常分类：AppError 存 code+message，其余存通用文案不泄 str(exc)。 P0-4 评审/正文 SSE 失败先发 error 事件，尾部 commit 包 try/except。 P1-4 max_version 加 FOR UPDATE 行锁消除 TOCTOU。 P1-5 scan_overdue 谓词下推 + 批量 UPDATE RETURNING。 P1-10 移除 OAuth user_code 日志。 P2 provider_deps 改调网关 build_adapter；accept_service Committable Protocol； CORS 白名单收窄；request_id 安全字符集白名单；stdlib 日志接管；读端点 404 校验； httpx timeout；测试用合法 Fernet key；类型化响应模型(JobResponse/DimensionEntry/ ReviewConflictView/selling_points)+路由 ErrorEnvelope responses（供 codegen）。
139 lines
4.8 KiB
Python
139 lines
4.8 KiB
Python
"""P0-1 原子性：`PUT /settings/providers` 多凭据一请求，中途失败整体回滚。
|
||
|
||
不变量：仓储写方法只 `flush()`，端点末尾**统一一次** `commit()`。若某条 upsert 中途
|
||
抛错，端点绝不到达 `commit()` → 请求 session 回滚，DB 不留半更新（无部分提交）。
|
||
|
||
用可注入 fake：store 在第 N 次 upsert 抛错；fake session 记录 `commit()` 调用次数。
|
||
断言：① 请求返回 500（兜底信封）；② `session.commit()` **从未被调用**（无半提交）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import uuid
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from cryptography.fernet import Fernet
|
||
from fastapi import FastAPI
|
||
from ww_api.services.credentials import StoredRouting
|
||
|
||
|
||
class _ExplodingStore:
|
||
"""凭据 store fake：前 N 次 upsert 成功，第 fail_at 次（1-based）抛错。
|
||
|
||
模拟「多凭据一请求」里第 K 条写入失败——验证端点不会半提交。
|
||
"""
|
||
|
||
def __init__(self, *, fail_at: int) -> None:
|
||
self._fail_at = fail_at
|
||
self._upsert_calls = 0
|
||
self.creds: dict[tuple[uuid.UUID, str], bytes] = {}
|
||
|
||
async def list_credentials(self, owner_id: uuid.UUID) -> list[Any]:
|
||
return []
|
||
|
||
async def list_routing(self) -> list[Any]:
|
||
return []
|
||
|
||
async def get_credential(self, owner_id: uuid.UUID, provider: str) -> Any:
|
||
return None
|
||
|
||
async def upsert_credential(
|
||
self, owner_id: uuid.UUID, provider: str, api_key_enc: bytes
|
||
) -> None:
|
||
self._upsert_calls += 1
|
||
if self._upsert_calls == self._fail_at:
|
||
raise RuntimeError("simulated mid-request write failure")
|
||
self.creds[(owner_id, provider)] = api_key_enc
|
||
|
||
async def upsert_oauth_credential(
|
||
self, owner_id: uuid.UUID, provider: str, oauth_enc: bytes
|
||
) -> None: # pragma: no cover - 不用于本测试
|
||
raise NotImplementedError
|
||
|
||
async def delete_credential(self, owner_id: uuid.UUID, provider: str) -> bool:
|
||
return False
|
||
|
||
async def upsert_routing(self, routing: StoredRouting) -> None:
|
||
self.creds[(uuid.UUID(int=0), routing.tier)] = b""
|
||
|
||
|
||
class _RecordingSession:
|
||
"""fake AsyncSession：只记 commit/rollback 调用次数（端点统一提交的探针）。"""
|
||
|
||
def __init__(self) -> None:
|
||
self.commits = 0
|
||
self.rollbacks = 0
|
||
|
||
async def commit(self) -> None:
|
||
self.commits += 1
|
||
|
||
async def rollback(self) -> None:
|
||
self.rollbacks += 1
|
||
|
||
|
||
def _build_app(store: _ExplodingStore, session: _RecordingSession) -> FastAPI:
|
||
os.environ.setdefault("CREDENTIAL_ENC_KEY", Fernet.generate_key().decode())
|
||
from ww_config import get_settings
|
||
|
||
get_settings.cache_clear()
|
||
from ww_api.main import create_app
|
||
from ww_api.services.provider_deps import get_credential_store
|
||
from ww_db import get_session
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_credential_store] = lambda: store
|
||
app.dependency_overrides[get_session] = lambda: session
|
||
return app
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_upsert_providers_rolls_back_on_mid_request_failure() -> None:
|
||
# Arrange：第 2 条凭据写入时抛错（前一条已 flush 到 session，但尚未 commit）。
|
||
store = _ExplodingStore(fail_at=2)
|
||
session = _RecordingSession()
|
||
app = _build_app(store, session)
|
||
|
||
payload = {
|
||
"credentials": [
|
||
{"provider": "deepseek", "api_key": "sk-a"},
|
||
{"provider": "kimi", "api_key": "sk-b"},
|
||
{"provider": "qwen", "api_key": "sk-c"},
|
||
]
|
||
}
|
||
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
resp = await client.put("/settings/providers", json=payload)
|
||
|
||
# Assert：兜底信封（catch-all → 500 INTERNAL，不回显原始异常），且**从未提交**（无半提交）。
|
||
assert resp.status_code == 500
|
||
assert resp.json()["error"]["code"] == "INTERNAL"
|
||
assert "simulated mid-request write failure" not in resp.text
|
||
assert session.commits == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_upsert_providers_commits_once_on_success() -> None:
|
||
# Arrange：全部成功（fail_at 远大于条数）。
|
||
store = _ExplodingStore(fail_at=99)
|
||
session = _RecordingSession()
|
||
app = _build_app(store, session)
|
||
|
||
payload = {
|
||
"credentials": [
|
||
{"provider": "deepseek", "api_key": "sk-a"},
|
||
{"provider": "kimi", "api_key": "sk-b"},
|
||
]
|
||
}
|
||
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
resp = await client.put("/settings/providers", json=payload)
|
||
|
||
# Assert：成功路径**恰好一次** commit（所有写入统一提交）。
|
||
assert resp.status_code == 200
|
||
assert session.commits == 1
|