ClaudeandClaude
25c8b039ac
- test_write_commit_persists：成功写操作被 db_session_middleware 提交 - test_write_rollback_on_validation_error：参数校验失败回滚，无脏数据 - test_operation_log_persists_on_success/failure：operation_log 独立会话落库，业务失败也记录 - 集成测试共 7 项，全部通过 Co-Authored-By: Claude <noreply@anthropic.com>
78 lines
3.3 KiB
Python
78 lines
3.3 KiB
Python
"""事务与操作日志集成测试：db_session_middleware 的 commit/rollback 分支、operation_log 独立落库。
|
||
|
||
需真实 MySQL/Redis 运行（默认标记 integration 跳过）。
|
||
"""
|
||
|
||
import pytest
|
||
import pymysql
|
||
|
||
from config import DB_HOST, DB_PORT, DB_DATABASE, DB_USERNAME, DB_PASSWORD
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
|
||
def _query_count(table, where, *args):
|
||
conn = pymysql.connect(host=DB_HOST, port=int(DB_PORT), user=DB_USERNAME,
|
||
password=DB_PASSWORD, database=DB_DATABASE)
|
||
cur = conn.cursor()
|
||
cur.execute(f"SELECT COUNT(*) FROM `{table}` WHERE {where}", args)
|
||
n = cur.fetchone()[0]
|
||
conn.close()
|
||
return n
|
||
|
||
|
||
def _execute(sql, *args):
|
||
conn = pymysql.connect(host=DB_HOST, port=int(DB_PORT), user=DB_USERNAME,
|
||
password=DB_PASSWORD, database=DB_DATABASE)
|
||
cur = conn.cursor()
|
||
cur.execute(sql, args)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
_VALID_DEPT = {"name": "测试部门T", "code": "TESTT", "type": 1, "pid": 0, "sort": 99, "status": 1}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_write_commit_persists(client, admin_headers):
|
||
"""业务成功的写操作应被 db_session_middleware 提交。"""
|
||
r = await client.post("/dept/add", headers=admin_headers, json=_VALID_DEPT)
|
||
assert r.json()["code"] == 0
|
||
assert _query_count("fastapi_dept", "name=%s", "测试部门T") >= 1
|
||
# 清理测试数据
|
||
_execute("DELETE FROM fastapi_dept WHERE name=%s", "测试部门T")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_write_rollback_on_validation_error(client, admin_headers):
|
||
"""参数校验失败（code!=0）时请求事务应回滚，不产生脏数据。"""
|
||
before = _query_count("fastapi_dept", "name=%s", "测试部门X")
|
||
r = await client.post("/dept/add", headers=admin_headers, json={"name": "测试部门X"}) # 缺必填字段
|
||
assert r.json()["code"] != 0
|
||
assert _query_count("fastapi_dept", "name=%s", "测试部门X") == before
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_operation_log_persists_on_success(client, admin_headers):
|
||
"""成功的写操作后，operation_log 应通过独立会话落库（标题正确）。"""
|
||
before = _query_count("fastapi_operation_log", "1=1")
|
||
r = await client.post("/dept/add", headers=admin_headers, json=_VALID_DEPT)
|
||
assert r.json()["code"] == 0
|
||
after = _query_count("fastapi_operation_log", "1=1")
|
||
assert after >= before + 1, "operation_log 未新增记录"
|
||
_execute("DELETE FROM fastapi_dept WHERE name=%s", "测试部门T")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_operation_log_persists_on_failure(client, admin_headers):
|
||
"""业务失败的写操作也应记录 operation_log（status=1），且独立会话不受请求回滚影响。"""
|
||
# 构造一个通过参数校验但业务失败的操作：部门编码重复（唯一性校验）
|
||
await client.post("/dept/add", headers=admin_headers, json=_VALID_DEPT)
|
||
before = _query_count("fastapi_operation_log", "1=1")
|
||
# 再次添加同名同编码 → 业务失败
|
||
r = await client.post("/dept/add", headers=admin_headers, json=_VALID_DEPT)
|
||
assert r.json()["code"] != 0
|
||
after = _query_count("fastapi_operation_log", "1=1")
|
||
assert after >= before + 1, "业务失败的写操作未记录 operation_log"
|
||
_execute("DELETE FROM fastapi_dept WHERE name=%s", "测试部门T")
|