5a3778a424
三处 create（业务/发货/回款）的 return _retry_create 是吞异常后正常 退出：外层审计把已回滚的首次尝试记成 success（假成功），重入整个 create_* 违反 audit_service「不支持嵌套」约定，且 _NO_RETRY 死常量 导致持续冲突时无界递归至 RecursionError。改为 audit 内 for 循环 （上限 _NO_RETRY），rollback 后重新锁行与取号，成功/最终失败各恰好 一条审计。新增集成用例锁定两条路径的日志增量。 Co-Authored-By: Claude Code <noreply@anthropic.com>
161 lines
6.2 KiB
Python
161 lines
6.2 KiB
Python
"""单据编号唯一键冲突重试的行为锁定（audit 上下文内循环重试）。
|
||
|
||
守：1062 兜底重试若以「吞异常后 return 重入整个 create_*」实现，会产生
|
||
三重缺陷 —— 外层审计把已回滚的首次尝试记成 success=True（假成功，与
|
||
audit_service.py「上下文内不得吞掉异常后正常退出」的约定相反）、重试再
|
||
开一层审计（违反「不支持嵌套」）、持续冲突时递归至 RecursionError。
|
||
修复口径：审计上下文内 for 循环重试（上限 _NO_RETRY），成功/最终失败
|
||
各恰好一条操作日志（以 POST 前的日志条数为基线断言增量）。
|
||
|
||
制造冲突的手段：monkeypatch ``business_service._generate_no`` 首次返回
|
||
库内已存在编号 → flush 撞唯一键 → 走重试分支。审计按 username 冗余溯
|
||
源（账号删除后 user_id 被 SET NULL），断言以基线增量免受既有数据干扰。
|
||
"""
|
||
|
||
import uuid
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import func, select
|
||
|
||
from app.core.constants import ACTION_CREATE, MODULE_BUSINESS
|
||
from app.core.db import get_session_factory
|
||
from app.main import app
|
||
from app.models import OperationLog
|
||
from app.services import business_service
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
_BUSINESSES = "/api/v1/business/businesses"
|
||
_CUSTOMERS = "/api/v1/base/customers"
|
||
|
||
|
||
@pytest.fixture
|
||
def quiet_client():
|
||
"""持续冲突用例依赖全局异常处理器产出 500 信封。
|
||
|
||
TestClient 默认对 ServerErrorMiddleware 的响应直接 re-raise
|
||
（raise_server_exceptions=True），会掩盖「原子处理器已正确收敛 500」
|
||
的行为本身 —— 审计断言反而无从谈起。
|
||
"""
|
||
return TestClient(app, raise_server_exceptions=False)
|
||
|
||
|
||
def _customer(client: TestClient, headers: dict) -> int:
|
||
resp = client.post(
|
||
_CUSTOMERS,
|
||
json={
|
||
"company_name": f"t_retry公司_{uuid.uuid4().hex[:8]}",
|
||
"owner_name": "张老板",
|
||
"customer_type": "personal",
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
return int(resp.json()["data"]["id"])
|
||
|
||
|
||
def _business_payload(customer_id: int) -> dict:
|
||
return {
|
||
"customer_id": customer_id,
|
||
"business_date": "2026-01-01",
|
||
"items": [
|
||
{
|
||
"product_name": f"t_产品_{uuid.uuid4().hex[:8]}",
|
||
"product_category": "municipal_water",
|
||
"quantity": "1.00",
|
||
"sale_price": "10.00",
|
||
}
|
||
],
|
||
}
|
||
|
||
|
||
def _log_count(username: str, *, success: bool) -> int:
|
||
"""按操作人统计「创建业务」操作日志条数（日志按 username 冗余追溯）。"""
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
stmt = (
|
||
select(func.count())
|
||
.select_from(OperationLog)
|
||
.where(
|
||
OperationLog.username == username,
|
||
OperationLog.module == MODULE_BUSINESS,
|
||
OperationLog.action == ACTION_CREATE,
|
||
OperationLog.summary.like("创建业务%"),
|
||
OperationLog.success == int(success),
|
||
)
|
||
)
|
||
return int(session.execute(stmt).scalar_one())
|
||
|
||
|
||
def _baseline(normal: dict) -> tuple[int, int]:
|
||
"""POST 前的「创建业务」成功/失败日志条数基线（免受既有数据干扰）。"""
|
||
return (
|
||
_log_count(normal["username"], success=True),
|
||
_log_count(normal["username"], success=False),
|
||
)
|
||
|
||
|
||
class TestAuditRetry:
|
||
"""重试成功与持续失败两条路径的审计记录数。"""
|
||
|
||
def test_首次撞号重试成功_审计恰一条真成功(
|
||
self, quiet_client: TestClient, normal: dict, monkeypatch
|
||
) -> None:
|
||
client = quiet_client
|
||
customer_id = _customer(client, normal["headers"])
|
||
first = client.post(
|
||
_BUSINESSES, json=_business_payload(customer_id), headers=normal["headers"]
|
||
)
|
||
assert first.status_code == 200, first.text
|
||
existing_no = first.json()["data"]["business_no"]
|
||
|
||
real_generate = business_service._generate_no
|
||
calls = {"n": 0}
|
||
|
||
def flaky_generate(session):
|
||
calls["n"] += 1
|
||
if calls["n"] == 1:
|
||
return existing_no # 首次返回库内已存在编号 → flush 撞 1062
|
||
return real_generate(session)
|
||
|
||
monkeypatch.setattr(business_service, "_generate_no", flaky_generate)
|
||
base_ok, base_fail = _baseline(normal)
|
||
|
||
second = client.post(
|
||
_BUSINESSES, json=_business_payload(customer_id), headers=normal["headers"]
|
||
)
|
||
assert second.status_code == 200, second.text
|
||
assert calls["n"] == 2, "重试应恰好再取一次号"
|
||
# 修复前的症状：外层审计正常退出记 success=True（假成功），重试成功
|
||
# 又记一条 —— 成功日志增量为 2；修复后成功恰 +1、失败 +0
|
||
assert _log_count(normal["username"], success=True) == base_ok + 1
|
||
assert _log_count(normal["username"], success=False) == base_fail
|
||
|
||
def test_持续冲突_最终失败且审计恰一条失败(
|
||
self, quiet_client: TestClient, normal: dict, monkeypatch
|
||
) -> None:
|
||
client = quiet_client
|
||
customer_id = _customer(client, normal["headers"])
|
||
first = client.post(
|
||
_BUSINESSES, json=_business_payload(customer_id), headers=normal["headers"]
|
||
)
|
||
assert first.status_code == 200, first.text
|
||
existing_no = first.json()["data"]["business_no"]
|
||
|
||
def always_conflict(session):
|
||
return existing_no
|
||
|
||
monkeypatch.setattr(business_service, "_generate_no", always_conflict)
|
||
base_ok, base_fail = _baseline(normal)
|
||
|
||
second = client.post(
|
||
_BUSINESSES, json=_business_payload(customer_id), headers=normal["headers"]
|
||
)
|
||
# 未捕获 IntegrityError 由全局异常兜底收敛 500
|
||
assert second.status_code == 500, second.text
|
||
# 修复前的症状:return _retry_create 无界递归直至 RecursionError,
|
||
# 每层各写一条记录；修复后重试有上限,最终失败仅一条失败日志
|
||
assert _log_count(normal["username"], success=False) == base_fail + 1
|
||
assert _log_count(normal["username"], success=True) == base_ok
|