feat(system): 完成后端系统管理（审计/角色/操作日志/日志查询）
- 操作日志审计机制：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>
This commit is contained in:
liuwenxinandClaude Code committed
1 parent
e22c4bcdc9
commit
c83287819a
32 files changed
+3158
-137
No files matched your search
@@ -0,0 +1,128 @@
|
||||
"""操作日志入库前处理单元测试。
|
||||
|
||||
守的是入库前的**截断与脱敏**：
|
||||
|
||||
- 截断：各列超长在 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 "�" 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
|
||||
@@ -0,0 +1,268 @@
|
||||
"""操作日志审计服务单元测试。
|
||||
|
||||
守的是 `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 "")
|
||||
@@ -0,0 +1,92 @@
|
||||
"""日志查询模型单元测试。
|
||||
|
||||
守的是**时间范围校验**与**时区归一**：起始晚于结束时必须拒绝 —— 放过去
|
||||
的话 SQL 条件互斥、查询恒为空，使用者会以为是"没有数据"而不是"条件写反了"；
|
||||
带时区的输入必须转换为本地 naive 时间 —— 不转换时 aware/naive 混用比较
|
||||
抛 TypeError（穿透成 500），或偏移被驱动静默丢弃（查询窗口偏 8 小时）。
|
||||
"""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.schemas.log import LoginLogQuery, OperationLogQuery
|
||||
|
||||
|
||||
class TestTimeRange:
|
||||
"""start_time ≤ end_time 校验（两类日志共用基类）。"""
|
||||
|
||||
def test_合法范围通过(self) -> None:
|
||||
query = OperationLogQuery(
|
||||
start_time=datetime(2026, 10, 1, 0, 0, 0), end_time=datetime(2026, 10, 10, 0, 0, 0)
|
||||
)
|
||||
assert query.start_time is not None
|
||||
|
||||
def test_起始晚于结束被拒(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
OperationLogQuery(start_time=datetime(2026, 10, 2), end_time=datetime(2026, 10, 1))
|
||||
|
||||
def test_登录日志同样校验(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
LoginLogQuery(start_time=datetime(2026, 10, 2), end_time=datetime(2026, 10, 1))
|
||||
|
||||
def test_边界相等通过(self) -> None:
|
||||
moment = datetime(2026, 10, 1, 12, 0, 0)
|
||||
assert OperationLogQuery(start_time=moment, end_time=moment).start_time == moment
|
||||
|
||||
def test_仅传一端不校验(self) -> None:
|
||||
assert OperationLogQuery(start_time=datetime(2026, 10, 1)).end_time is None
|
||||
assert OperationLogQuery(end_time=datetime(2026, 10, 1)).start_time is None
|
||||
|
||||
|
||||
class TestQueryFields:
|
||||
"""筛选字段边界。"""
|
||||
|
||||
def test_成功标记只能为0或1(self) -> None:
|
||||
assert OperationLogQuery(success=0).success == 0
|
||||
with pytest.raises(ValidationError):
|
||||
OperationLogQuery(success=2)
|
||||
|
||||
def test_分页别名与偏移(self) -> None:
|
||||
query = OperationLogQuery(pageNum=2, pageSize=50)
|
||||
assert query.offset == 50
|
||||
|
||||
def test_操作日志筛选字段(self) -> None:
|
||||
query = OperationLogQuery(module="用户管理", action="删除", username="adm")
|
||||
assert (query.module, query.action, query.username) == ("用户管理", "删除", "adm")
|
||||
|
||||
def test_登录日志按IP精确筛选(self) -> None:
|
||||
assert LoginLogQuery(ip="10.0.0.1").ip == "10.0.0.1"
|
||||
|
||||
|
||||
class TestTimezoneNormalization:
|
||||
"""时区归一。
|
||||
|
||||
回归背景：混合 aware/naive 比较曾抛 TypeError 穿透成 500；aware 直传
|
||||
曾被驱动静默丢弃偏移（查询窗口整体偏 8 小时且无提示）。
|
||||
"""
|
||||
|
||||
def test_utc后缀被转换为本地时间(self) -> None:
|
||||
"""Z 后缀 = UTC 零点 = 东八区本地 08:00（与库列的本地时间口径对齐）。"""
|
||||
query = OperationLogQuery(start_time="2026-10-01T00:00:00Z")
|
||||
assert query.start_time == datetime(2026, 10, 1, 8, 0, 0)
|
||||
|
||||
def test_混合aware与naive不再抛比较错误(self) -> None:
|
||||
query = OperationLogQuery(
|
||||
start_time=datetime(2026, 10, 1, 0, 0, 0),
|
||||
end_time=datetime(2026, 10, 2, 0, 0, 0, tzinfo=UTC),
|
||||
)
|
||||
assert query.end_time is not None
|
||||
assert query.end_time.tzinfo is None # 已摘除
|
||||
|
||||
def test_归一后范围校验仍按本地口径生效(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
OperationLogQuery(
|
||||
start_time="2026-10-02T00:00:00Z", # 本地 10-02 08:00
|
||||
end_time="2026-10-01 12:00:00", # 本地 10-01 12:00，早于起始
|
||||
)
|
||||
|
||||
def test_naive时间视为本地原样保留(self) -> None:
|
||||
moment = datetime(2026, 10, 1, 12, 0, 0)
|
||||
assert OperationLogQuery(start_time=moment).start_time == moment
|
||||
@@ -0,0 +1,98 @@
|
||||
"""角色请求模型单元测试。
|
||||
|
||||
守的是**角色编码的格式红线**：编码会进入授权判断（`ROLE_ADMIN` 按编码
|
||||
比对）与操作日志，必须限定为小写字母开头的小写/数字/下划线组合 ——
|
||||
放开大写或中文会让对照与检索困难，超长则与列宽（32）不符。
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.schemas.role import RoleCreateIn, RoleQuery, RoleStatusIn, RoleUpdateIn
|
||||
|
||||
|
||||
def _create(**overrides: object) -> RoleCreateIn:
|
||||
base: dict[str, object] = {"role_code": "finance", "role_name": "财务"}
|
||||
base.update(overrides)
|
||||
return RoleCreateIn(**base)
|
||||
|
||||
|
||||
class TestRoleCodePattern:
|
||||
"""编码规则：^[a-z][a-z0-9_]{1,31}$"""
|
||||
|
||||
def test_合法编码通过(self) -> None:
|
||||
assert _create().role_code == "finance"
|
||||
|
||||
def test_下划线与数字通过(self) -> None:
|
||||
assert _create(role_code="sales_01").role_code == "sales_01"
|
||||
|
||||
def test_大写字母被拒(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
_create(role_code="Finance")
|
||||
|
||||
def test_中文被拒(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
_create(role_code="财务")
|
||||
|
||||
def test_数字开头被拒(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
_create(role_code="1abc")
|
||||
|
||||
def test_下划线开头被拒(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
_create(role_code="_abc")
|
||||
|
||||
def test_单字符被拒(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
_create(role_code="a")
|
||||
|
||||
def test_超长被拒(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
_create(role_code="a" * 33)
|
||||
|
||||
def test_首尾空白被自动去除(self) -> None:
|
||||
assert _create(role_code=" finance ").role_code == "finance"
|
||||
|
||||
|
||||
class TestRoleUpdate:
|
||||
"""编辑模型：不含 role_code。"""
|
||||
|
||||
def test_编辑不接受role_code(self) -> None:
|
||||
"""编码是授权判定锚点，编辑接口有意不接收；多余字段被忽略而非报错
|
||||
（前端表单可能整份回填提交）。"""
|
||||
payload = RoleUpdateIn(role_name="财务二组", role_code="hacked")
|
||||
assert not hasattr(payload, "role_code")
|
||||
|
||||
def test_名称为空被拒(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
RoleUpdateIn(role_name="")
|
||||
|
||||
def test_排序负数被拒(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
RoleUpdateIn(role_name="财务", sort_order=-1)
|
||||
|
||||
|
||||
class TestRoleStatusAndQuery:
|
||||
"""状态与查询模型。"""
|
||||
|
||||
def test_状态只能为0或1(self) -> None:
|
||||
assert RoleStatusIn(status=0).status == 0
|
||||
assert RoleStatusIn(status=1).status == 1
|
||||
with pytest.raises(ValidationError):
|
||||
RoleStatusIn(status=2)
|
||||
|
||||
def test_查询默认分页(self) -> None:
|
||||
query = RoleQuery()
|
||||
assert query.page_num == 1
|
||||
assert query.page_size == 10
|
||||
assert query.keyword is None
|
||||
assert query.status is None
|
||||
|
||||
def test_分页别名照常工作(self) -> None:
|
||||
"""pageNum/pageSize 是前端契约（PageQuery 的 alias）。"""
|
||||
query = RoleQuery(pageNum=3, pageSize=5)
|
||||
assert query.offset == 10
|
||||
|
||||
def test_关键词超长被拒(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
RoleQuery(keyword="a" * 65)
|
||||
Reference in new issue
Block a user