c83287819a
- 操作日志审计机制：audit 上下文（成功与失败都记、独立提交、操作人标量 固化、commit 后查询前置约定），用户/角色全部写操作接入；request_data 按键名的结构化脱敏 + UTF-8 字节边界截断 - 角色管理：CRUD 与启停；admin 禁删禁停、有绑定禁删（FOR UPDATE 锁行）、 编码创建后不可改、停用角色不可再分配 - 日志查询：操作/登录两端点，列表与详情两段式（详情含已脱敏请求参数）； 时间参数统一转本地 naive（修混用时区 500 与偏移丢弃） - 用户管理：补齐 reset_password 操作人、编辑空串归一 NULL、并发唯一键兜底 400 - 测试 301 例全绿（单元 223 / 集成 78） Co-Authored-By: Claude Code <noreply@anthropic.com>
269 lines
9.8 KiB
Python
269 lines
9.8 KiB
Python
"""操作日志审计服务单元测试。
|
||
|
||
守的是 `audit()` 上下文管理器的语义：
|
||
|
||
- 正常退出记成功、抛异常记失败，**失败记录必须先生成再让异常继续传播**
|
||
- 失败路径必须先 `rollback` 再写日志 —— 业务异常若来自 IntegrityError 等
|
||
数据库错误，不回滚时 session 处于待回滚状态，写日志会抛 PendingRollbackError，
|
||
最该留下的失败记录反而被静默丢掉
|
||
- 操作人信息在进入上下文时**固化为标量** —— 失败路径 rollback 会过期 ORM
|
||
对象，再读属性会触发额外查询，账号并发被删时甚至 ObjectDeletedError
|
||
令失败审计静默丢失
|
||
- 审计自身的任何失败都不得外抛（"审计可丢，业务不可断"）
|
||
"""
|
||
|
||
import logging
|
||
from datetime import datetime
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from app.core.exceptions import BadRequestError
|
||
from app.models import OperationLog
|
||
from app.services.audit_service import AuditRecord, RequestMeta, audit
|
||
|
||
|
||
class _StubSession:
|
||
"""最小会话桩：记录调用顺序与写入的日志对象，可注入失败。"""
|
||
|
||
def __init__(self) -> None:
|
||
self.events: list[str] = []
|
||
self.logs: list[OperationLog] = []
|
||
self.fail_add = False
|
||
self.fail_commit = False
|
||
# audit 入口会巡检未提交变更（嵌套调用防护）
|
||
self.new: list = []
|
||
self.dirty: list = []
|
||
self.deleted: list = []
|
||
|
||
def add(self, obj: object) -> None:
|
||
if self.fail_add:
|
||
raise RuntimeError("模拟写入失败")
|
||
self.events.append("add")
|
||
self.logs.append(obj) # type: ignore[arg-type]
|
||
|
||
def flush(self) -> None:
|
||
self.events.append("flush")
|
||
|
||
def commit(self) -> None:
|
||
if self.fail_commit:
|
||
raise RuntimeError("模拟提交失败")
|
||
self.events.append("commit")
|
||
|
||
def rollback(self) -> None:
|
||
self.events.append("rollback")
|
||
|
||
|
||
class _FrozenOperator:
|
||
"""只允许读取固定轮数属性的操作人桩。
|
||
|
||
模拟「rollback 后 ORM 对象过期」：若失败路径仍读取 operator 属性，
|
||
真实环境会触发额外 SELECT（且账号并发被删时抛 ObjectDeletedError），
|
||
此处直接以异常暴露该访问。audit 入口读 id + username 恰为 2 次。
|
||
"""
|
||
|
||
def __init__(self, max_reads: int = 2) -> None:
|
||
self._max_reads = max_reads
|
||
self._reads = 0
|
||
|
||
def _touch(self) -> None:
|
||
self._reads += 1
|
||
if self._reads > self._max_reads:
|
||
raise RuntimeError("进入审计后不应再访问操作人 ORM 属性")
|
||
|
||
@property
|
||
def id(self) -> int:
|
||
self._touch()
|
||
return 7
|
||
|
||
@property
|
||
def username(self) -> str:
|
||
self._touch()
|
||
return "admin"
|
||
|
||
|
||
def _operator() -> SimpleNamespace:
|
||
"""模拟已认证用户（audit 只读取 id 与 username 两个属性）。"""
|
||
return SimpleNamespace(id=7, username="admin")
|
||
|
||
|
||
def _record(**overrides: object) -> AuditRecord:
|
||
base: dict = {"module": "用户管理", "action": "新增", "summary": "新增用户 zhangsan"}
|
||
base.update(overrides)
|
||
return AuditRecord(**base)
|
||
|
||
|
||
def _audit(
|
||
stub: _StubSession,
|
||
*,
|
||
request_data: dict | None = None,
|
||
meta: RequestMeta | None = None,
|
||
):
|
||
return audit(
|
||
stub, # type: ignore[arg-type]
|
||
operator=_operator(),
|
||
record=_record(request_data=request_data),
|
||
meta=meta,
|
||
)
|
||
|
||
|
||
class TestSuccessPath:
|
||
"""正常退出：业务完成后记成功并独立提交。"""
|
||
|
||
def test_成功路径写入一条成功日志(self) -> None:
|
||
stub = _StubSession()
|
||
with _audit(stub):
|
||
pass
|
||
assert len(stub.logs) == 1
|
||
log = stub.logs[0]
|
||
assert log.success == 1
|
||
assert log.error_message is None
|
||
assert log.module == "用户管理"
|
||
assert log.action == "新增"
|
||
assert log.username == "admin"
|
||
assert log.user_id == 7
|
||
|
||
def test_日志提交发生在业务逻辑之后(self) -> None:
|
||
"""审计是业务的"后置动作"：业务先 commit，日志再独立 commit。"""
|
||
stub = _StubSession()
|
||
with _audit(stub):
|
||
stub.events.append("business")
|
||
assert stub.events == ["business", "add", "flush", "commit"]
|
||
|
||
def test_请求元信息与耗时被写入(self) -> None:
|
||
stub = _StubSession()
|
||
meta = RequestMeta(
|
||
ip="10.0.0.8", user_agent="pytest-agent", method="POST", path="/api/v1/system/users"
|
||
)
|
||
with _audit(stub, meta=meta):
|
||
pass
|
||
log = stub.logs[0]
|
||
assert log.method == "POST"
|
||
assert log.path == "/api/v1/system/users"
|
||
assert log.ip == "10.0.0.8"
|
||
assert log.user_agent == "pytest-agent"
|
||
assert log.duration_ms is not None
|
||
assert log.duration_ms >= 0
|
||
|
||
def test_请求参数经脱敏后写入(self) -> None:
|
||
stub = _StubSession()
|
||
with _audit(stub, request_data={"username": "zhangsan", "password": "Secret@2026"}):
|
||
pass
|
||
request_data = stub.logs[0].request_data or ""
|
||
assert "Secret@2026" not in request_data
|
||
assert "zhangsan" in request_data
|
||
|
||
|
||
class TestFailurePath:
|
||
"""业务异常：先回滚、记失败、异常原样重抛。"""
|
||
|
||
def test_失败时原异常原样重抛(self) -> None:
|
||
stub = _StubSession()
|
||
with pytest.raises(BadRequestError, match="角色编码无效"), _audit(stub):
|
||
raise BadRequestError("角色编码无效")
|
||
|
||
def test_失败记录在回滚之后写入(self) -> None:
|
||
"""顺序断言：rollback 必须早于 add。颠倒时，来自数据库异常的失败
|
||
记录会因 session 待回滚而抛 PendingRollbackError，整条丢失。"""
|
||
stub = _StubSession()
|
||
with pytest.raises(BadRequestError), _audit(stub):
|
||
stub.events.append("business")
|
||
raise BadRequestError("用户名重复")
|
||
assert stub.events == ["business", "rollback", "add", "flush", "commit"]
|
||
log = stub.logs[0]
|
||
assert log.success == 0
|
||
assert log.error_message == "用户名重复"
|
||
|
||
def test_未预期异常不泄漏内部信息(self) -> None:
|
||
"""未预期异常的堆栈由全局处理器记入应用日志；审计表只落通用文案。"""
|
||
stub = _StubSession()
|
||
with pytest.raises(ValueError, match="内部实现细节"), _audit(stub):
|
||
raise ValueError("内部实现细节")
|
||
assert stub.logs[0].error_message == "服务端异常，请稍后重试"
|
||
|
||
|
||
class TestOperatorFrozen:
|
||
"""操作人标量在进入上下文时固化，之后（尤其 rollback 后）不得再触碰 ORM 对象。"""
|
||
|
||
def test_成功路径只读取一轮操作人属性(self) -> None:
|
||
stub = _StubSession()
|
||
with audit(stub, operator=_FrozenOperator(), record=_record()): # type: ignore[arg-type]
|
||
pass
|
||
assert stub.logs[0].username == "admin"
|
||
|
||
def test_失败路径不再触碰操作人对象(self) -> None:
|
||
"""回归：此前失败路径在 rollback 之后读 operator.id/username ——
|
||
真实环境会触发额外 SELECT，账号被并发删除时 ObjectDeletedError
|
||
还会被兜底吞掉，最该留痕的失败审计静默丢失。"""
|
||
stub = _StubSession()
|
||
with (
|
||
pytest.raises(BadRequestError, match="用户不存在"),
|
||
audit(
|
||
stub, # type: ignore[arg-type]
|
||
operator=_FrozenOperator(),
|
||
record=_record(action="删除", summary="删除用户 x"),
|
||
),
|
||
):
|
||
raise BadRequestError("用户不存在")
|
||
assert stub.logs[0].user_id == 7
|
||
assert stub.logs[0].username == "admin"
|
||
|
||
|
||
class TestDirtySessionGuard:
|
||
"""进入审计时存在未提交变更 → 记 warning（防嵌套调用造成半提交）。"""
|
||
|
||
def test_脏会话触发警告(self, caplog: pytest.LogCaptureFixture) -> None:
|
||
stub = _StubSession()
|
||
stub.dirty.append(object())
|
||
with caplog.at_level(logging.WARNING), _audit(stub):
|
||
pass
|
||
assert any("未提交的变更" in message for message in caplog.messages)
|
||
|
||
def test_干净会话不告警(self, caplog: pytest.LogCaptureFixture) -> None:
|
||
stub = _StubSession()
|
||
with caplog.at_level(logging.WARNING), _audit(stub):
|
||
pass
|
||
assert not caplog.messages
|
||
|
||
|
||
class TestResilience:
|
||
"""审计失败不得影响业务。"""
|
||
|
||
def test_写入失败不外抛且业务继续(self) -> None:
|
||
stub = _StubSession()
|
||
stub.fail_add = True
|
||
reached = False
|
||
with _audit(stub):
|
||
reached = True
|
||
assert reached is True # 审计写入失败不打断业务流
|
||
|
||
def test_提交失败不外抛(self) -> None:
|
||
stub = _StubSession()
|
||
stub.fail_commit = True
|
||
with _audit(stub):
|
||
pass # 正常退出，不抛异常
|
||
|
||
def test_业务失败时审计失败不以审计异常替换原异常(self) -> None:
|
||
"""业务异常是用户看到的原因，不能因审计自身失败而被替换。"""
|
||
stub = _StubSession()
|
||
stub.fail_add = True
|
||
with pytest.raises(BadRequestError, match="业务错误"), _audit(stub):
|
||
raise BadRequestError("业务错误")
|
||
|
||
|
||
class TestSerializationEdge:
|
||
"""审计参数的序列化边界（细节覆盖见 test_audit_serialization）。"""
|
||
|
||
def test_请求参数可为None(self) -> None:
|
||
stub = _StubSession()
|
||
with _audit(stub, request_data=None):
|
||
pass
|
||
assert stub.logs[0].request_data is None
|
||
|
||
def test_不可JSON序列化的值被降级为字符串(self) -> None:
|
||
"""如 datetime 等对象经 default=str 降级，序列化不应抛异常。"""
|
||
stub = _StubSession()
|
||
with _audit(stub, request_data={"at": datetime(2026, 10, 10, 12, 0, 0)}):
|
||
pass
|
||
assert "2026-10-10" in (stub.logs[0].request_data or "")
|