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>
129 lines
5.5 KiB
Python
129 lines
5.5 KiB
Python
"""操作日志入库前处理单元测试。
|
||
|
||
守的是入库前的**截断与脱敏**：
|
||
|
||
- 截断：各列超长在 MySQL 严格模式下抛 1406，而「审计失败不影响业务」的约定
|
||
会把该异常吞掉 —— 整条日志静默消失；request_data 尤甚，中文一个字符
|
||
占 3 字节，必须按 **UTF-8 字节边界**切，否则会切出乱码残字
|
||
- 脱敏：密码是最容易从日志泄漏的凭据，request_data 入库前必经 scrub
|
||
"""
|
||
|
||
import json
|
||
from datetime import datetime
|
||
|
||
from app.models.operation_log import (
|
||
ERROR_MESSAGE_MAX_LEN,
|
||
REQUEST_DATA_MAX_BYTES,
|
||
SUMMARY_MAX_LEN,
|
||
)
|
||
from app.repositories.operation_log_repository import _clip, _serialize_request_data
|
||
|
||
|
||
class TestClip:
|
||
"""按字符数截断（长度常量与数据库列宽均按字符定义）。"""
|
||
|
||
def test_超长截断至上限(self) -> None:
|
||
result = _clip("x" * (SUMMARY_MAX_LEN + 50), SUMMARY_MAX_LEN)
|
||
assert result is not None
|
||
assert len(result) == SUMMARY_MAX_LEN
|
||
|
||
def test_未超长原样返回(self) -> None:
|
||
assert _clip("正常长度", SUMMARY_MAX_LEN) == "正常长度"
|
||
|
||
def test_空值返回None(self) -> None:
|
||
assert _clip(None, ERROR_MESSAGE_MAX_LEN) is None
|
||
assert _clip("", ERROR_MESSAGE_MAX_LEN) is None
|
||
|
||
def test_中文按字符数截断不切字(self) -> None:
|
||
"""str 切片天然按字符，不会切出半个汉字。"""
|
||
result = _clip("中" * 100, 10)
|
||
assert result == "中" * 10
|
||
|
||
|
||
class TestSerializeRequestData:
|
||
"""序列化：JSON 化 → 脱敏 → 字节边界截断。"""
|
||
|
||
def test_常规参数原样序列化(self) -> None:
|
||
text = _serialize_request_data({"status": 0, "roles": ["user"]})
|
||
assert text is not None
|
||
assert json.loads(text) == {"status": 0, "roles": ["user"]}
|
||
|
||
def test_空值返回None(self) -> None:
|
||
assert _serialize_request_data(None) is None
|
||
assert _serialize_request_data({}) is None
|
||
|
||
def test_密码值被抹除而键名保留(self) -> None:
|
||
"""键名保留便于审计阅读"提交过密码字段"，值必须抹掉。"""
|
||
for payload in ({"password": "S3cret@2026"}, {"new_password": "S3cret@2026"}):
|
||
text = _serialize_request_data(payload) or ""
|
||
assert "S3cret@2026" not in text
|
||
assert "***" in text
|
||
|
||
def test_手机号被脱敏(self) -> None:
|
||
text = _serialize_request_data({"phone": "13812345678"}) or ""
|
||
assert "13812345678" not in text
|
||
assert "138****5678" in text
|
||
|
||
def test_超长中文按字节边界截断且无残字(self) -> None:
|
||
"""75000 字节（25000 个汉字）必然超限：
|
||
若按字符截断则字节数三倍超标；若按字节硬切可能切出半个字符。
|
||
正确结果应是原始 JSON 串的**干净前缀**。"""
|
||
payload = {"remark": "中" * 25000}
|
||
text = _serialize_request_data(payload)
|
||
assert text is not None
|
||
assert len(text.encode("utf-8")) <= REQUEST_DATA_MAX_BYTES
|
||
assert "<EFBFBD>" not in text # 无替换字符
|
||
raw = json.dumps(payload, ensure_ascii=False)
|
||
assert raw.startswith(text)
|
||
|
||
def test_未超长的中文完整保留(self) -> None:
|
||
payload = {"remark": "中" * 100}
|
||
text = _serialize_request_data(payload)
|
||
assert text is not None
|
||
assert json.loads(text) == payload
|
||
|
||
def test_不可JSON序列化的值降级为字符串(self) -> None:
|
||
text = _serialize_request_data({"at": datetime(2026, 10, 10, 12, 0, 0)})
|
||
assert text is not None
|
||
assert "2026-10-10" in text
|
||
|
||
|
||
class TestStructuredSanitization:
|
||
"""结构化抹除 —— 按键名整值替换，不依赖文本正则。
|
||
|
||
回归背景：此前用文本正则脱敏，有三种逃逸形态 —— 值以空白开头时
|
||
整体匹配失败（**完全不脱敏**）；值含空格/逗号时只抹首段；值含引号时
|
||
既残留明文又破坏 JSON 结构。密码校验只查字节数、前后端都不 trim，
|
||
这些形态都是正常可用密码。
|
||
"""
|
||
|
||
def test_前导空白密码被完全抹除(self) -> None:
|
||
"""回归：此前 `" abc12345"` 原样明文入库。"""
|
||
text = _serialize_request_data({"new_password": " abc12345"}) or ""
|
||
assert "abc12345" not in text
|
||
assert json.loads(text) == {"new_password": "***"}
|
||
|
||
def test_含空格逗号引号均整体抹除且JSON完整(self) -> None:
|
||
for password in ("abc 12345", "abc,def", 'abc"def', "abc&def"):
|
||
text = _serialize_request_data({"password": password}) or ""
|
||
assert password not in text
|
||
assert json.loads(text) == {"password": "***"}
|
||
|
||
def test_嵌套结构与键名大小写(self) -> None:
|
||
text = (
|
||
_serialize_request_data({"user": {"New_Password": "s3"}, "list": [{"TOKEN": "t"}]})
|
||
or ""
|
||
)
|
||
assert "s3" not in text
|
||
assert json.loads(text) == {"user": {"New_Password": "***"}, "list": [{"TOKEN": "***"}]}
|
||
|
||
def test_非敏感键不受影响(self) -> None:
|
||
text = _serialize_request_data({"keyword": "a_b", "remark": "正常备注"}) or ""
|
||
assert json.loads(text) == {"keyword": "a_b", "remark": "正常备注"}
|
||
|
||
def test_孤立代理字符不再抛错(self) -> None:
|
||
"""回归：直接 encode('utf-8') 遇 \\ud800 抛 UnicodeEncodeError，
|
||
该请求的审计记录会被兜底逻辑吞掉、静默丢失。"""
|
||
text = _serialize_request_data({"username": "a\ud800b"})
|
||
assert text is not None
|