merge: 同步 origin/qyqy_develop（风控扫描、知识检索、客服 Agent、协商等）
冲突处理：均为双方各自新增，按并集保留—— - .gitignore：本地 data/logs 忽略项 + 远端 .dsh-drop/ - app/core/config.py：offsite/promotion 与 risk_scan 配置项并存 - app/service/agent/bootstrap.py：场外/推介/风控/NL2SQL/知识检索 工具与 Agent 全部注册 - app/worker/__main__.py：场外邮件 Worker 接线 + runtime 关系服务/投影清理注入并存 收尾：新增 20260911_merge_risk_heads 收敛迁移双 head；按 docs/21 生成 config/jwt/dev 开发密钥。
This commit is contained in:
@@ -0,0 +1,178 @@
|
||||
"""风控写接口的幂等语义（`docs/05` §5.1、§5.2；`docs/25` P3 #22）。
|
||||
|
||||
两级覆盖：
|
||||
|
||||
1. `ApiTransactionService.execute_in` 本身 —— 同键同正文回放原响应且**不重复执行**、
|
||||
同键不同正文返回 `409 IDEMPOTENCY_CONFLICT`、缺键直接拒绝；
|
||||
2. Controller 接线 —— 重复 POST 同一个风控处置端点时，Action Service 只被调用一次。
|
||||
|
||||
刻意不去驱动真实状态机：处置动作会改预警状态，测试不该污染演示数据。这里用计数替身
|
||||
验证"第二次请求没有落到业务逻辑上"，这才是幂等要保证的事情。
|
||||
"""
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from sqlalchemy import delete
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api.controllers import risk as risk_controller
|
||||
from app.api.dependencies.auth import build_request_context
|
||||
from app.api.dependencies.database import get_session
|
||||
from app.core.contracts import RequestContext
|
||||
from app.core.errors import IdempotencyConflictError, ValidationAgentError
|
||||
from app.infrastructure.db import SessionFactory
|
||||
from app.main import app
|
||||
from app.repository.platform_repository import PlatformRepository
|
||||
from app.service.api_transaction_service import ApiTransactionService
|
||||
|
||||
SCOPE = "POST /api/v1/risk/alerts/ALERT-IDEM-TEST/acknowledgements"
|
||||
EXCLUSION_SCOPE = "POST /api/v1/risk/alerts/ALERT-IDEM-TEST/exclusions"
|
||||
ACK_PATH = "/api/v1/risk/alerts/ALERT-IDEM-TEST/acknowledgements"
|
||||
EXCLUSION_PATH = "/api/v1/risk/alerts/ALERT-IDEM-TEST/exclusions"
|
||||
|
||||
|
||||
async def override_context() -> RequestContext:
|
||||
return RequestContext(
|
||||
user_id="990000002",
|
||||
trace_id=str(uuid4()),
|
||||
roles=("risk_operator",),
|
||||
permissions=("risk:alert:read", "risk:alert:write"),
|
||||
data_scope="all",
|
||||
)
|
||||
|
||||
|
||||
async def override_session() -> AsyncIterator[AsyncSession]:
|
||||
async with SessionFactory() as session:
|
||||
yield session
|
||||
|
||||
|
||||
async def purge(*keys: str) -> None:
|
||||
async with SessionFactory() as session, session.begin():
|
||||
table = await PlatformRepository(session).table("api_request_receipt")
|
||||
for key in keys:
|
||||
await session.execute(delete(table).where(table.c.idempotency_key == key))
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_in_replays_without_running_action_twice() -> None:
|
||||
key = f"risk-idem-{uuid4()}"
|
||||
context = await override_context()
|
||||
calls: list[int] = []
|
||||
|
||||
async def action(_session: AsyncSession) -> dict[str, Any]:
|
||||
calls.append(1)
|
||||
return {"alert_no": "ALERT-IDEM-TEST", "status": "待处理"}
|
||||
|
||||
try:
|
||||
async with SessionFactory() as session:
|
||||
service = ApiTransactionService()
|
||||
first = await service.execute_in(session, context, SCOPE, key, {}, action)
|
||||
second = await service.execute_in(session, context, SCOPE, key, {}, action)
|
||||
|
||||
assert first == second == {"alert_no": "ALERT-IDEM-TEST", "status": "待处理"}
|
||||
assert len(calls) == 1, "重复请求必须回放 response_json，而不是再次执行 action"
|
||||
finally:
|
||||
await purge(key)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_in_conflicts_on_same_key_with_different_body() -> None:
|
||||
key = f"risk-idem-{uuid4()}"
|
||||
context = await override_context()
|
||||
|
||||
async def action(_session: AsyncSession) -> dict[str, Any]:
|
||||
return {"ok": True}
|
||||
|
||||
try:
|
||||
async with SessionFactory() as session:
|
||||
service = ApiTransactionService()
|
||||
await service.execute_in(session, context, SCOPE, key, {"reason": "第一次"}, action)
|
||||
with pytest.raises(IdempotencyConflictError):
|
||||
await service.execute_in(
|
||||
session, context, SCOPE, key, {"reason": "第二次"}, action
|
||||
)
|
||||
finally:
|
||||
await purge(key)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_in_rejects_missing_or_short_key() -> None:
|
||||
context = await override_context()
|
||||
|
||||
async def action(_session: AsyncSession) -> dict[str, Any]:
|
||||
raise AssertionError("缺键时不应执行 action")
|
||||
|
||||
async with SessionFactory() as session:
|
||||
service = ApiTransactionService()
|
||||
for bad in (None, "too-short", "带中文字符的-key-1234567890"):
|
||||
with pytest.raises(ValidationAgentError):
|
||||
await service.execute_in(session, context, SCOPE, bad, {}, action)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_repeated_risk_post_calls_action_service_once(monkeypatch) -> None:
|
||||
"""端到端：同一 `Idempotency-Key` 重复 POST 只触发一次处置逻辑。
|
||||
|
||||
用 `httpx.ASGITransport` 而不是 `TestClient`：后者自建事件循环，测试结束后的
|
||||
`SessionFactory` 清理会落在另一个循环上，连接池析构时报
|
||||
`AttributeError: 'NoneType' object has no attribute 'send'`。
|
||||
"""
|
||||
calls: list[str] = []
|
||||
|
||||
class CountingActionService:
|
||||
def __init__(self, _session: Any) -> None:
|
||||
pass
|
||||
|
||||
async def acknowledge(self, alert_no: str, _context: RequestContext) -> dict[str, Any]:
|
||||
calls.append(alert_no)
|
||||
return {"alert_no": alert_no, "status": "待处理", "ack_status": "已确认"}
|
||||
|
||||
async def exclude(
|
||||
self, alert_no: str, reason: str, _context: RequestContext
|
||||
) -> dict[str, Any]:
|
||||
calls.append(f"exclude:{reason}")
|
||||
return {"alert_no": alert_no, "status": "已排除", "handle_result": reason}
|
||||
|
||||
monkeypatch.setattr(risk_controller, "RiskActionService", CountingActionService)
|
||||
ack_key = f"risk-ack-{uuid4()}"
|
||||
exclude_key = f"risk-exclude-{uuid4()}"
|
||||
app.dependency_overrides[build_request_context] = override_context
|
||||
app.dependency_overrides[get_session] = override_session
|
||||
try:
|
||||
transport = httpx.ASGITransport(app=app)
|
||||
async with httpx.AsyncClient(
|
||||
transport=transport, base_url="http://testserver"
|
||||
) as client:
|
||||
first = await client.post(ACK_PATH, headers={"Idempotency-Key": ack_key})
|
||||
replay = await client.post(ACK_PATH, headers={"Idempotency-Key": ack_key})
|
||||
excluded = await client.post(
|
||||
EXCLUSION_PATH,
|
||||
json={"reason": "客户本人确认"},
|
||||
headers={"Idempotency-Key": exclude_key},
|
||||
)
|
||||
conflict = await client.post(
|
||||
EXCLUSION_PATH,
|
||||
json={"reason": "换了理由"},
|
||||
headers={"Idempotency-Key": exclude_key},
|
||||
)
|
||||
|
||||
assert first.status_code == 200
|
||||
assert replay.status_code == 200
|
||||
# 只比 `data`：`meta.trace_id` 标识的是**本次**请求，重放也必须换一个新的。
|
||||
assert replay.json()["data"] == first.json()["data"]
|
||||
assert excluded.status_code == 200
|
||||
assert conflict.status_code == 409
|
||||
assert conflict.json()["error"]["code"] == "IDEMPOTENCY_CONFLICT"
|
||||
assert calls == ["ALERT-IDEM-TEST", "exclude:客户本人确认"]
|
||||
finally:
|
||||
app.dependency_overrides.pop(build_request_context, None)
|
||||
app.dependency_overrides.pop(get_session, None)
|
||||
await purge(ack_key, exclude_key)
|
||||
Reference in New Issue
Block a user