"""操作日志审计服务单元测试。 守的是 `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 "")