chore: 初始化项目结构与登录鉴权模块

后端 FastAPI 六层结构(api/services/repositories/models/schemas/core),
统一响应信封、全局异常、日志脱敏;6 张表经 Alembic 迁移建库,SQL 留存
db/migration;登录鉴权含图形验证码(Redis GETDEL 防重放)、JWT、登录限流、
登出黑名单(TTL=令牌剩余有效期)、登录日志。

前端 Vue3 + TS + Element Plus 骨架,含登录页、路由守卫、请求封装与 Vite 代理。

工程配置:uv 锁定 Python 3.12、ruff 检查与格式化、editorconfig、gitattributes。

测试 184 项(单元 155 / 集成 29)全绿,浏览器端到端 8 项通过。

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
liuwenxinandClaude Code committed 2026-10-09 18:46:36 +08:00
commit 273ebf4657
101 files changed
+8725

No files matched your search

+98
View File
@@ -0,0 +1,98 @@
# 后端测试
> AI摘要:本项目后端的测试用例清单与运行记录。按「基础设施层随实现即时测、业务模块整模块完成时统一测」分层组织。当前 184 个用例全绿(单元 155 / 集成 29)。
## 一、测试约定
### 编写时机(分层对待)
- `core/` 等**基础设施**随实现即时补测 —— 它们是被业务复用多次的地基,越早锁定越省事
- **业务模块**(系统管理、基础数据管理、业务数据管理、绩效工资核算、报表统计)在**整个模块功能完成时**统一编写用例并执行,不在模块做到一半时零散补测
### 目录二分
| 目录 | 依赖 | 用途 |
| --- | --- | --- |
| `tests/unit/` | 无(不碰 Docker、不连库) | 纯函数与元数据验证,秒级跑完,开发中可高频执行 |
| `tests/integration/` | 完整应用实例或 MySQL/Redis 容器 | 接口级、跨产物一致性验证 |
单元测试里的 Redis 依赖通过 `fake_redis` 夹具(fakeredis)替换,因此验证码 GETDEL、限流计数、黑名单 TTL 这些逻辑无需起容器即可覆盖。
### 注释要求
- 测试类用中文 docstring 说明「这组用例守的是什么」——即**如果不测会出什么问题**
- 用例名描述场景与预期(`test_中文按字节而非字符计算边界`)
- 关键断言补注释说明为何该断言重要,尤其是防回归的用例
### 运行方式
```bash
cd backend
uv run pytest tests/unit -q # 纯单元,无需容器,秒级
uv run pytest -q # 全部(集成部分需容器在跑)
uv run pytest tests/integration -q # 仅集成
```
## 二、用例清单
### 基础设施层(core)
| 测试文件 | 守什么 | 用例数 | 关键覆盖点 |
| --- | --- | :---: | --- |
| `unit/test_models.py` | 建表规范红线 | 49 | 表名小写蛇形复数;每列中文注释**且确实输出到 DDL**;`uk_`/`idx_`/`fk_` 命名前缀;ENGINE/CHARSET 显式指定;`ON UPDATE CURRENT_TIMESTAMP` 存在(`server_onupdate` 不会生成该子句,是 SQLAlchemy 的行为陷阱);日志表反向断言不含该子句 |
| `unit/test_logging.py` | 日志脱敏与格式化 | 16 | 密码/令牌/Bearer/手机号/身份证五类脱敏;**脱敏必须后置于 %-格式化**(早期实现在 filter 阶段改写 `record.msg`,抹掉 `%s` 占位符导致 logging 抛 TypeError、整条日志丢失);setup_logging 幂等 |
| `unit/test_config.py` | 配置加载与安全边界 | 14 | 密钥强度 fail-fast;JWT ≤120 分钟、验证码 ≤300 秒的硬约束;密码含 `@ : / #` 时 URL 构造不破坏;渲染 URL 默认隐藏密码 |
| `unit/test_security.py` | 密码哈希 | 11 | BCrypt 格式与定长 60;**72 字节边界**(24 个汉字正好、25 个超限);超长密码校验时返回 False 而非抛异常(防侧信道);哈希串损坏不炸 |
| `unit/test_response.py` | 统一响应契约 | 11 | 字段固定为 `code/message/data`;分页序列化为规范要求的驼峰 `list/total/pageNum/pageSize` |
| `unit/test_exception_handlers.py` | 异常转统一信封 | 5 | 校验错误 422→400 并压成单行中文;HTTP 状态码与 body 的 `code` 必须一致 |
### 系统管理 · 登录鉴权
| 测试文件 | 守什么 | 用例数 | 关键覆盖点 |
| --- | --- | :---: | --- |
| `unit/test_jwt.py` | 令牌有效期语义与版本机制 | 12 | `exp` 基于 **UTC** 而非本地时间(混算会产生 8 小时偏差,表现为「刚登录就过期」或「过期了还能用」);过期/篡改签名/**alg=none 攻击**/畸形字符串一律收敛为 401 而非 500;jti 唯一 |
| `unit/test_captcha.py` | 验证码**一次性语义** | 12 | GETDEL 后二次使用必然失败(防重放);**校验失败也消费**(否则可用一个验证码反复试密码);字符集剔除 `0O1Il`;每次渲染含随机干扰不可比对;TTL 300 秒 |
| `unit/test_login_limiter.py` | 阈值与**滑动窗口** | 9 | 达阈值即锁;每次失败刷新 TTL(否则「4 次失败 + 等 14 分钟 + 1 次失败」可绕过锁定);按用户名隔离,一个账号被攻击不牵连他人;登录成功清除计数 |
| `unit/test_token_blacklist.py` | 黑名单 **TTL 语义** | 7 | TTL 等于令牌剩余有效期(写死固定值会导致「登出 N 分钟后又活了」,不设过期则 Redis 无限增长);已过期令牌不写入;jti 互不串号 |
| `unit/test_auth_schemas.py` | 密码**按字节**校验 | 9 | 24 个汉字 = 72 字节正好、25 个被拒 —— 按字符数校验(如 `max_length=20`)根本拦不住,会让登录接口 500 |
| `integration/test_auth_flow.py` | 整条链路**串联正确性** | 16 | 验证码→登录→持令牌访问→登出失效;验证码重放被拒;失败原因如实落库且**日志表无任何密码字段**;账号不存在与密码错误文案完全相同(防枚举);禁用账号与 `token_version` 提升均使旧令牌立即失效 |
### 应用骨架与交付产物
| 测试文件 | 守什么 | 用例数 | 关键覆盖点 |
| --- | --- | :---: | --- |
| `integration/test_health.py` | 统一信封与全局异常 | 7 | `/api/health` 始终 200 且依赖状态如实上报;404/405 返回统一信封而非框架默认 `detail` 数组;`/api/docs` 可访问 |
| `integration/test_schema_sync.py` | DDL 不漂移 | 6 | `db/migration/V1__*.sql` 与 `db/init/init-dev.sql` 的**表集合与列集合逐表比对**——两份 DDL 手工同步迟早漂移,且漂移后果隐蔽(开发库结构与生产不一致,上线才发现) |
## 三、运行记录
| 日期 | 范围 | 用例数 | 结果 | 备注 |
| --- | --- | :---: | --- | --- |
| 2026-10-09 | 全部 | 120 | ✅ 通过(4.99s) | 阶段 1 骨架 + 阶段 2 数据库完成后首次全量运行 |
| 2026-10-09 | 全部 | 119 | ✅ 通过(5.01s) | ORM 由异步改回同步后回归;净减 1 例(`sync_database_url` 独立配置已合并) |
| 2026-10-09 | 全部 | 184 | ✅ 通过(25.46s) | **登录鉴权模块完成**后统一编写并执行该模块测试套件(新增 65 例)。集成用例需容器在跑 |
## 四、测试发现并修复的缺陷
| 日期 | 缺陷 | 影响 | 处理 |
| --- | --- | --- | --- |
| 2026-10-09 | 日志脱敏在 `logging.Filter` 阶段改写 `record.msg` | 抹掉 `%s` 占位符 → logging 内部 `msg % args` 抛 TypeError → **整条日志丢失**(比不脱敏更危险) | 改为 `SensitiveFormatter` 在格式化**之后**脱敏;补回归用例 |
| 2026-10-09 | `PasswordHash.recommended()` 在 pwdlib 0.3.x 推荐 Argon2 而非 BCrypt | 密码算法与规范不符,且随库版本变化——同一批密码可能混用两种算法 | 改为显式 `PasswordHash((BcryptHasher(),))` 钉死算法 |
| 2026-10-09 | 模型未设置表级 `comment`,`update_time` 用 `server_onupdate` | 违反规范红线;`server_onupdate` 只是 ORM 标记,**不生成 `ON UPDATE CURRENT_TIMESTAMP` 子句**,更新时间永不刷新 | 补表注释;改用 `text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP")` |
| 2026-10-09 | `pymysql` 未带 `[rsa]` extra,`cryptography` 缺失 | MySQL 8.0 默认 `caching_sha2_password`,容器重启后首次连接走全量认证时抛 `RuntimeError`,且报错在建连阶段易被误读为「密码错误」 | 依赖改为 `pymysql[rsa]` |
| 2026-10-09 | `setex` 在 redis-py 8 已弃用 | 规范明确**禁用弃用 API**;弃用接口在未来大版本会被移除 | 验证码与黑名单改用 `set(key, value, ex=…)` |
| 2026-10-09 | 测试中 TTL 断言写死等值 | 属**测试自身缺陷**:Redis 的 TTL 以毫秒精度存绝对过期时刻、返回时向下取整,刚写入就查询必然可能少 1 秒——会变成偶发失败的闪烁测试 | 断言改为容忍 1 秒偏差,并注明原因 |
| 2026-10-09 | 验证码沿用 `captcha` 库默认渲染 | 随机彩色字符 + 重噪点,与界面视觉体系冲突;该库最后更新于 2025-03,内部还引用了 Pillow 的旧别名 | 改为自绘(`core/captcha_image.py`):配色与界面统一,难度由逐字旋转、抖动、干扰曲线与噪点保证;仅沿用其内置字体文件 |
| 2026-10-09 | 噪点用 `ellipse` 绘制 | 3×3 包围盒画出的椭圆恰好填充中心与四个边中点,形状就是「+」,形成规则的十字纹理——既显眼又容易被整体滤波抹掉,起不到干扰作用 | 改用单像素点与 2×2 方块 |
| 2026-10-09 | `tests/unit/test_jwt.py` 两处赋值了未使用的 `settings`(ruff F841) | 变量未被引用,属无效代码 | 删除赋值;该文件的 `get_settings` 导入仍被其他用例使用 |
## 五、待补用例
| 模块 | 状态 | 说明 |
| --- | --- | --- |
| 系统管理 · 用户与角色管理 | 待编写 | 需求 2.5.1 的用户增删改查、密码重置、启用/禁用、角色分配 — 该模块功能尚未实现 |
| 系统管理 · 操作日志 | 待编写 | 除登录日志外的操作审计(当前仅登录日志落库) |
| 基础数据管理 | 未开始 | 依赖 T+ 对接方式确认 |
| 业务数据管理 | 未开始 | 依赖需求澄清(回款天数口径等 10 项) |
| 绩效工资核算 | 未开始 | 依赖需求澄清;金额精度需专项用例(`Fraction` 与 1/90 日斜率) |
| 报表统计 | 未开始 | 依赖报表口径确认 |
+1
View File
@@ -0,0 +1 @@
"""后端测试包。"""
+38
View File
@@ -0,0 +1,38 @@
"""测试共用夹具。"""
import pytest
from fastapi.testclient import TestClient
@pytest.fixture(scope="session")
def client() -> TestClient:
"""构造应用测试客户端(会话级复用,避免重复构建应用)。
注意:测试环境未必有 MySQL/Redis 在跑,健康检查会如实返回 false,
因此断言应聚焦于响应结构而非依赖的连通性。
"""
from app.main import app
with TestClient(app) as test_client:
yield test_client
@pytest.fixture
def fake_redis(monkeypatch: pytest.MonkeyPatch):
"""把 Redis 客户端替换为进程内假实现。
验证码的 GETDEL、限流的 INCR/EXPIRE/TTL、黑名单的 SETEX——都只用到
基础字符串命令,fakeredis 能完整覆盖,因此这些逻辑无需起容器即可测试,
单测得以保持秒级。
替换的是 `app.core.redis_client` 的模块级单例,各 service 通过
`get_redis()` 取用,故一处替换即全局生效。
"""
import fakeredis
from app.core import redis_client
fake = fakeredis.FakeRedis(decode_responses=True)
monkeypatch.setattr(redis_client, "_client", fake)
yield fake
fake.close()
+5
View File
@@ -0,0 +1,5 @@
"""需要构造完整应用或依赖外部服务(MySQL/Redis)的测试。
`tests/unit/` 下的用例必须能在没有 Docker 的机器上秒跑;
凡是要连容器或用例起整个 FastAPI 应用的,归入本目录。
"""
+271
View File
@@ -0,0 +1,271 @@
"""登录链路端到端测试(需要 MySQL 与 Redis)。
守的是整条链路的**串联正确性**:验证码 → 登录 → 持令牌访问 → 登出失效。
各环节细节已由单元测试覆盖,这里验证它们接起来是否真的可用 ——
尤其是「判定顺序」「失败时日志是否真的落库」「禁用与提版能否立即生效」
这类只有串联运行才会暴露的问题。
运行前置:``docker compose -f docker/docker-compose.dev.yml up -d``
"""
import uuid
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import delete, select
from app.core.constants import REDIS_KEY_CAPTCHA, REDIS_KEY_LOGIN_FAIL
from app.core.db import get_session_factory
from app.core.redis_client import get_redis
from app.core.security import hash_password
from app.models import LoginLog, User
pytestmark = pytest.mark.integration
TEST_PASSWORD = "TestPwd@2026"
@pytest.fixture
def test_user():
"""创建独立测试账号,用后即删。
不复用 admin:测试会触发账号锁定、提升 token_version 等操作,
污染真实账号会让后续手工验证产生困惑。
"""
factory = get_session_factory()
username = f"t_{uuid.uuid4().hex[:8]}"
with factory() as session:
user = User(
username=username,
password=hash_password(TEST_PASSWORD),
real_name="测试账号",
status=1,
)
session.add(user)
session.commit()
user_id = user.id
yield {"id": user_id, "username": username}
with factory() as session:
session.execute(delete(LoginLog).where(LoginLog.username == username))
session.execute(delete(User).where(User.id == user_id)) # user_roles 级联删除
session.commit()
get_redis().delete(REDIS_KEY_LOGIN_FAIL.format(username=username))
def _fetch_captcha(client: TestClient) -> tuple[str, str]:
"""取验证码及其明文答案。
直接从 Redis 读取答案 —— **仅测试可用**,生产环境当然不能如此,
这里是为了让自动化用例不必做图像识别。
"""
data = client.get("/api/v1/auth/captcha").json()["data"]
captcha_id = data["captcha_id"]
answer = get_redis().get(REDIS_KEY_CAPTCHA.format(captcha_id=captcha_id))
return captcha_id, answer
def _login(client: TestClient, username: str, password: str, **overrides):
"""发起一次登录,默认使用新鲜验证码。"""
captcha_id, code = _fetch_captcha(client)
payload = {
"username": username,
"password": password,
"captcha_id": captcha_id,
"captcha_code": code,
}
payload.update(overrides)
return client.post("/api/v1/auth/login", json=payload)
def _bearer(token: str) -> dict[str, str]:
return {"Authorization": f"Bearer {token}"}
class TestHappyPath:
"""正常链路。"""
def test_验证码到登出的完整链路(self, client: TestClient, test_user: dict) -> None:
captcha_id, code = _fetch_captcha(client)
login = client.post(
"/api/v1/auth/login",
json={
"username": test_user["username"],
"password": TEST_PASSWORD,
"captcha_id": captcha_id,
"captcha_code": code,
},
)
assert login.status_code == 200
body = login.json()
assert body["code"] == 200
token = body["data"]["access_token"]
assert body["data"]["token_type"] == "Bearer"
assert body["data"]["user"]["username"] == test_user["username"]
me = client.get("/api/v1/auth/me", headers=_bearer(token))
assert me.json()["data"]["username"] == test_user["username"]
assert client.post("/api/v1/auth/logout", headers=_bearer(token)).json()["code"] == 200
after = client.get("/api/v1/auth/me", headers=_bearer(token))
assert after.status_code == 401
assert after.json()["code"] == 401
def test_登录成功后清除失败计数(self, client: TestClient, test_user: dict) -> None:
username = test_user["username"]
for _ in range(3):
_login(client, username, "WrongPwd@1")
assert get_redis().exists(REDIS_KEY_LOGIN_FAIL.format(username=username)) == 1
assert _login(client, username, TEST_PASSWORD).status_code == 200
assert get_redis().exists(REDIS_KEY_LOGIN_FAIL.format(username=username)) == 0
class TestCaptcha:
"""验证码环节。"""
def test_同一验证码不可重放(self, client: TestClient, test_user: dict) -> None:
captcha_id, code = _fetch_captcha(client)
payload = {
"username": test_user["username"],
"password": TEST_PASSWORD,
"captcha_id": captcha_id,
"captcha_code": code,
}
assert client.post("/api/v1/auth/login", json=payload).status_code == 200
replay = client.post("/api/v1/auth/login", json=payload)
assert replay.status_code == 400
assert "验证码" in replay.json()["message"]
def test_验证码错误时拒绝登录(self, client: TestClient, test_user: dict) -> None:
response = _login(client, test_user["username"], TEST_PASSWORD, captcha_code="ZZZZ")
assert response.status_code == 400
assert "验证码" in response.json()["message"]
def test_验证码错误不消耗失败次数(self, client: TestClient, test_user: dict) -> None:
"""验证码与密码是两类失败,混在一起计数会让用户被误锁。"""
username = test_user["username"]
for _ in range(5):
_login(client, username, TEST_PASSWORD, captcha_code="ZZZZ")
# 之后用正确验证码仍可正常登录
assert _login(client, username, TEST_PASSWORD).status_code == 200
class TestCredentials:
"""凭据校验。"""
def test_密码错误被拒(self, client: TestClient, test_user: dict) -> None:
response = _login(client, test_user["username"], "WrongPwd@1")
assert response.status_code == 400
def test_账号不存在与密码错误的提示完全相同(self, client: TestClient) -> None:
"""文案不一致会形成账号枚举信道,攻击者可据此筛出真实用户名。"""
missing = _login(client, f"nobody_{uuid.uuid4().hex[:8]}", "Whatever@1")
assert missing.status_code == 400
assert missing.json()["message"] == "用户名或密码错误"
def test_无令牌访问受保护接口返回401(self, client: TestClient) -> None:
assert client.get("/api/v1/auth/me").status_code == 401
def test_伪造令牌返回401(self, client: TestClient) -> None:
response = client.get("/api/v1/auth/me", headers=_bearer("not.a.real.token"))
assert response.status_code == 401
assert response.json()["code"] == 401
class TestAccountState:
"""账号状态与令牌失效。"""
def test_禁用账号无法登录(self, client: TestClient, test_user: dict) -> None:
with get_session_factory()() as session:
user = session.get(User, test_user["id"])
user.status = 0
session.commit()
response = _login(client, test_user["username"], TEST_PASSWORD)
assert response.status_code == 403
assert "禁用" in response.json()["message"]
def test_禁用后已签发的令牌立即失效(self, client: TestClient, test_user: dict) -> None:
token = _login(client, test_user["username"], TEST_PASSWORD).json()["data"]["access_token"]
assert client.get("/api/v1/auth/me", headers=_bearer(token)).status_code == 200
with get_session_factory()() as session:
user = session.get(User, test_user["id"])
user.status = 0
session.commit()
assert client.get("/api/v1/auth/me", headers=_bearer(token)).status_code == 401
def test_提升token_version使旧令牌立即失效(self, client: TestClient, test_user: dict) -> None:
"""这是「离职即失效」的实现手段,且**不依赖 Redis**。"""
token = _login(client, test_user["username"], TEST_PASSWORD).json()["data"]["access_token"]
assert client.get("/api/v1/auth/me", headers=_bearer(token)).status_code == 200
with get_session_factory()() as session:
user = session.get(User, test_user["id"])
user.token_version += 1 # 相当于改密 / 强制下线
session.commit()
assert client.get("/api/v1/auth/me", headers=_bearer(token)).status_code == 401
class TestRateLimit:
"""登录限流。"""
def test_连续失败达阈值后正确密码也被拒(self, client: TestClient, test_user: dict) -> None:
username = test_user["username"]
for _ in range(5):
_login(client, username, "WrongPwd@1")
response = _login(client, username, TEST_PASSWORD)
assert response.status_code == 403
assert "锁定" in response.json()["message"]
def test_锁定期内失败计数未被继续放大(self, client: TestClient, test_user: dict) -> None:
"""锁定后仍不断尝试不应无限刷新 TTL——否则会被用来永久锁死他人账号。"""
username = test_user["username"]
for _ in range(5):
_login(client, username, "WrongPwd@1")
ttl_before = get_redis().ttl(REDIS_KEY_LOGIN_FAIL.format(username=username))
for _ in range(3):
_login(client, username, TEST_PASSWORD)
ttl_after = get_redis().ttl(REDIS_KEY_LOGIN_FAIL.format(username=username))
assert ttl_after <= ttl_before
class TestLoginLogs:
"""登录日志落库。"""
def _logs(self, username: str) -> list[LoginLog]:
with get_session_factory()() as session:
return list(
session.execute(
select(LoginLog).where(LoginLog.username == username).order_by(LoginLog.id)
).scalars()
)
def test_成功与失败均落库且记录失败原因(self, client: TestClient, test_user: dict) -> None:
username = test_user["username"]
_login(client, username, TEST_PASSWORD)
_login(client, username, "WrongPwd@1")
_login(client, username, TEST_PASSWORD, captcha_code="ZZZZ")
reasons = [(log.success, log.fail_reason) for log in self._logs(username)]
assert (1, None) in reasons
assert (0, "用户名或密码错误") in reasons
assert (0, "验证码错误") in reasons
def test_日志不含任何密码字段(self, client: TestClient, test_user: dict) -> None:
"""日志是最容易泄漏凭据的地方,这条断言防止日后有人「顺手」加字段。"""
username = test_user["username"]
_login(client, username, TEST_PASSWORD)
_login(client, username, "WrongPwd@1")
columns = {column.name for column in LoginLog.__table__.columns}
assert not (columns & {"password", "passwd", "pwd", "credential"})
for log in self._logs(username):
assert TEST_PASSWORD not in str(log.__dict__)
+55
View File
@@ -0,0 +1,55 @@
"""应用级 HTTP 测试:统一响应格式与全局异常处理。
归在 integration 下是因为需要构造完整 FastAPI 应用;
但用例本身不要求 MySQL/Redis 在跑 —— 健康检查会把依赖状态如实体现在
data 中,接口始终返回 200。
"""
from fastapi.testclient import TestClient
class TestHealthEndpoint:
"""健康检查接口。"""
def test_返回统一响应格式(self, client: TestClient) -> None:
response = client.get("/api/health")
assert response.status_code == 200
body = response.json()
assert body["code"] == 200
assert body["message"] == "success"
assert body["data"]["app"] is True
def test_上报数据库与Redis状态为布尔值(self, client: TestClient) -> None:
data = client.get("/api/health").json()["data"]
assert isinstance(data["db"], bool)
assert isinstance(data["redis"], bool)
def test_不要求认证(self, client: TestClient) -> None:
assert client.get("/api/health").status_code == 200
class TestGlobalExceptionHandling:
"""全局异常处理器:不得把框架默认的 detail 数组泄漏给前端。"""
def test_未知路由返回统一格式(self, client: TestClient) -> None:
response = client.get("/api/definitely-not-exist")
assert response.status_code == 404
body = response.json()
assert body["code"] == 404
assert "detail" not in body
def test_方法不允许返回统一格式(self, client: TestClient) -> None:
response = client.post("/api/health")
assert response.status_code == 405
assert response.json()["code"] == 405
class TestApiDocs:
"""接口文档(规范第 10 节要求提供)。文档挂在 /api 下,便于经 Vite 代理访问。"""
def test_swagger文档可访问(self, client: TestClient) -> None:
assert client.get("/api/docs").status_code == 200
def test_openapi结构可访问(self, client: TestClient) -> None:
schema = client.get("/api/openapi.json").json()
assert "/api/health" in schema["paths"]
@@ -0,0 +1,89 @@
"""DDL 漂移检测。
结构定义同时存在于两处:
- `db/migration/V1__*.sql` —— 事实来源,生产迁移用
- `db/init/init-dev.sql` —— 开发环境重置脚本,Docker 首次启动唯一会自动执行的位置
两份 DDL 手工同步迟早会漂移,且后果隐蔽:开发库的表结构与迁移脚本不一致,
新人克隆下来建出的库和生产不一样,等到上线才发现。
本测试把「靠自觉」变成「可自动拦截」,成本极低。
注:本文件不需要容器,放在 integration 下是因为它校验的是交付产物而非应用代码。
"""
import re
from pathlib import Path
import pytest
PROJECT_ROOT = Path(__file__).resolve().parents[3]
MIGRATION_FILE = PROJECT_ROOT / "db" / "migration" / "V1__init_rbac_and_login_logs.sql"
INIT_DEV_FILE = PROJECT_ROOT / "db" / "init" / "init-dev.sql"
_CREATE_TABLE = re.compile(r"CREATE TABLE\s+`?(\w+)`?\s*\((.*?)\n\)", re.DOTALL)
_COLUMN_LINE = re.compile(r"^\s{4}`?(\w+)`?\s", re.MULTILINE)
# 这些开头的是约束定义,不是列定义
_NON_COLUMN_KEYWORDS = {"CONSTRAINT", "PRIMARY", "UNIQUE", "FOREIGN", "KEY", "INDEX"}
def _parse_tables(sql: str) -> dict[str, set[str]]:
"""从 SQL 文本中提取 {表名: {列名集合}}。"""
tables: dict[str, set[str]] = {}
for table_name, body in _CREATE_TABLE.findall(sql):
tables[table_name] = {
column
for column in _COLUMN_LINE.findall(body)
if column.upper() not in _NON_COLUMN_KEYWORDS
}
return tables
@pytest.fixture(scope="module")
def migration_tables() -> dict[str, set[str]]:
return _parse_tables(MIGRATION_FILE.read_text(encoding="utf-8"))
@pytest.fixture(scope="module")
def init_dev_tables() -> dict[str, set[str]]:
return _parse_tables(INIT_DEV_FILE.read_text(encoding="utf-8"))
class TestSchemaSync:
"""两份 DDL 的结构一致性。"""
def test_文件存在(self) -> None:
assert MIGRATION_FILE.is_file(), f"缺少迁移脚本:{MIGRATION_FILE}"
assert INIT_DEV_FILE.is_file(), f"缺少开发初始化脚本:{INIT_DEV_FILE}"
def test_表集合一致(self, migration_tables: dict, init_dev_tables: dict) -> None:
assert set(init_dev_tables) == set(migration_tables), (
"两份 DDL 的表集合不一致 —— 表结构变更时需同时维护 "
"db/migration/V{n}__*.sql 与 db/init/init-dev.sql"
)
def test_每张表的列集合一致(self, migration_tables: dict, init_dev_tables: dict) -> None:
mismatched = {
table: {
"仅迁移脚本有": sorted(migration_tables[table] - init_dev_tables[table]),
"仅初始化脚本有": sorted(init_dev_tables[table] - migration_tables[table]),
}
for table in set(migration_tables) & set(init_dev_tables)
if migration_tables[table] != init_dev_tables[table]
}
assert not mismatched, f"以下表的列不一致:{mismatched}"
class TestSeedData:
"""字典数据在两处应保持同步。"""
@pytest.mark.parametrize("role_code", ["admin", "user"])
def test_角色初始化数据在两处都存在(self, role_code: str) -> None:
migration_sql = MIGRATION_FILE.read_text(encoding="utf-8")
init_sql = INIT_DEV_FILE.read_text(encoding="utf-8")
assert f"'{role_code}'" in migration_sql
assert f"'{role_code}'" in init_sql
def test_开发脚本标记了迁移版本(self) -> None:
"""不写 alembic_version 的话,在开发库上执行 upgrade head 会重复建表而失败。"""
init_sql = INIT_DEV_FILE.read_text(encoding="utf-8")
assert "INSERT INTO alembic_version" in init_sql
+1
View File
@@ -0,0 +1 @@
"""不依赖外部服务的单元测试。"""
+71
View File
@@ -0,0 +1,71 @@
"""认证请求模型单元测试。
守的是:**密码按字节数校验**。
bcrypt 5.0 起对超过 72 字节的密码直接抛 ValueError(此前是静默截断),
而中文一个字符占 3 字节 —— 只按字符数校验(如 max_length=20)根本拦不住,
24 个汉字就会让哈希函数抛异常、登录接口返回 500。
"""
import pytest
from pydantic import ValidationError
from app.core.constants import CAPTCHA_LENGTH, PASSWORD_MAX_BYTES
from app.schemas.auth import LoginIn
def _payload(**overrides: object) -> dict[str, object]:
base: dict[str, object] = {
"username": "admin",
"password": "Pwd@12345",
"captcha_id": "a" * 32,
"captcha_code": "A" * CAPTCHA_LENGTH,
}
base.update(overrides)
return base
class TestPasswordByteLimit:
"""72 字节边界 —— 与 bcrypt 算法上限严格对齐。"""
def test_合法密码通过(self) -> None:
assert LoginIn(**_payload()).password == "Pwd@12345"
def test_恰好72字节ASCII通过(self) -> None:
password = "a" * PASSWORD_MAX_BYTES
assert LoginIn(**_payload(password=password)).password == password
def test_超过72字节ASCII被拒(self) -> None:
with pytest.raises(ValidationError):
LoginIn(**_payload(password="a" * (PASSWORD_MAX_BYTES + 1)))
def test_24个汉字恰好72字节通过(self) -> None:
"""一个汉字 3 字节:24 个正好等于上限。"""
password = "中" * 24
assert len(password.encode("utf-8")) == PASSWORD_MAX_BYTES
LoginIn(**_payload(password=password))
def test_25个汉字被拒(self) -> None:
"""若只按字符数校验,25 个字符远低于常见的 max_length=128,
会一路放行到哈希函数才炸成 500。"""
with pytest.raises(ValidationError):
LoginIn(**_payload(password="中" * 25))
class TestFieldConstraints:
"""其余字段的边界。"""
def test_验证码位数不符被拒(self) -> None:
with pytest.raises(ValidationError):
LoginIn(**_payload(captcha_code="AB"))
def test_用户名为空被拒(self) -> None:
with pytest.raises(ValidationError):
LoginIn(**_payload(username=""))
def test_用户名超长被拒(self) -> None:
with pytest.raises(ValidationError):
LoginIn(**_payload(username="a" * 65))
def test_首尾空白被自动去除(self) -> None:
"""用户复制粘贴账号时常带空格,自动去除比报错友好。"""
assert LoginIn(**_payload(username=" admin ")).username == "admin"
+91
View File
@@ -0,0 +1,91 @@
"""图形验证码单元测试。
守的是:**验证码的一次性语义**。若 GETDEL 退化为「先 GET 再 DEL」,
两条命令之间会出现竞态窗口,攻击者可用同一验证码并发重复提交 ——
而这个问题在正常使用下完全看不出来。
"""
import base64
from app.core.captcha_image import generate_text, render_png_base64
from app.core.constants import CAPTCHA_CHARSET, CAPTCHA_LENGTH, REDIS_KEY_CAPTCHA
from app.services import captcha_service
PNG_MAGIC = b"\x89PNG"
class TestCaptchaText:
"""验证码文本生成。"""
def test_长度固定(self) -> None:
assert len(generate_text()) == CAPTCHA_LENGTH
def test_字符集不含易混淆字符(self) -> None:
"""0/O、1/I/l 在图片上难以区分,会显著提高用户输错率。"""
text = "".join(generate_text() for _ in range(200))
assert set(text) <= set(CAPTCHA_CHARSET)
assert not (set(text) & set("0O1Il"))
def test_具备随机性(self) -> None:
assert len({generate_text() for _ in range(50)}) > 40
class TestCaptchaImage:
"""图片渲染。"""
def test_输出为合法PNG的base64(self) -> None:
raw = base64.b64decode(render_png_base64("AB3D"))
assert raw.startswith(PNG_MAGIC)
def test_不含data前缀(self) -> None:
"""返回纯 base64,由前端自行拼接 data:image/png;base64, ——
服务端不替调用方决定呈现方式。"""
assert not render_png_base64("AB3D").startswith("data:")
def test_同一文本两次渲染结果不同(self) -> None:
"""每次绘制含随机干扰线与噪点,防止通过图片比对反推答案。"""
assert render_png_base64("AB3D") != render_png_base64("AB3D")
class TestCaptchaService:
"""签发与校验(用假 Redis)。"""
def test_签发后可用正确验证码通过(self, fake_redis) -> None:
captcha_id, image, expires_in = captcha_service.issue()
assert expires_in == 300
assert base64.b64decode(image).startswith(PNG_MAGIC)
answer = fake_redis.get(REDIS_KEY_CAPTCHA.format(captcha_id=captcha_id))
assert captcha_service.verify_and_consume(captcha_id, answer) is True
def test_同一验证码二次使用必然失败(self, fake_redis) -> None:
"""GETDEL 一次性语义的核心断言 —— 防重放。"""
captcha_id, _, _ = captcha_service.issue()
answer = fake_redis.get(REDIS_KEY_CAPTCHA.format(captcha_id=captcha_id))
assert captcha_service.verify_and_consume(captcha_id, answer) is True
assert captcha_service.verify_and_consume(captcha_id, answer) is False
def test_校验失败后验证码同样被消费(self, fake_redis) -> None:
"""失败也要消费:否则攻击者可拿一个验证码反复试密码。"""
captcha_id, _, _ = captcha_service.issue()
answer = fake_redis.get(REDIS_KEY_CAPTCHA.format(captcha_id=captcha_id))
assert captcha_service.verify_and_consume(captcha_id, "XXXX") is False
assert captcha_service.verify_and_consume(captcha_id, answer) is False
def test_不存在的验证码返回False(self, fake_redis) -> None:
assert captcha_service.verify_and_consume("nonexistent-id", "AB3D") is False
def test_校验时忽略大小写与首尾空格(self, fake_redis) -> None:
"""用户从图片抄写时容易带上空格;答案统一按大写存储。"""
captcha_id, _, _ = captcha_service.issue()
answer = fake_redis.get(REDIS_KEY_CAPTCHA.format(captcha_id=captcha_id))
assert captcha_service.verify_and_consume(captcha_id, f" {answer.lower()} ") is True
def test_有效期300秒符合规范(self, fake_redis) -> None:
"""规范第 6 节要求验证码 300 秒。
容忍 1 秒偏差:Redis 的 TTL 以毫秒精度存绝对过期时刻,
命令返回时向下取整,刚写入就查询必然是 N 或 N-1。
"""
captcha_id, _, _ = captcha_service.issue()
ttl = fake_redis.ttl(REDIS_KEY_CAPTCHA.format(captcha_id=captcha_id))
assert ttl in (300, 299)
+83
View File
@@ -0,0 +1,83 @@
"""配置模块单元测试。
用 `_env_file=None` 绕过 backend/.env,使断言不受开发者本地配置影响。
"""
import pytest
from pydantic import ValidationError
from app.core.config import Settings
VALID_SECRET = "x" * 32
def _settings(**overrides: object) -> Settings:
"""构造一个不读取 .env 的配置实例。"""
params: dict[str, object] = {"_env_file": None, "app_secret_key": VALID_SECRET}
params.update(overrides)
return Settings(**params) # type: ignore[arg-type]
class TestSecretKey:
"""密钥强度校验。"""
def test_缺失密钥时拒绝构造(self) -> None:
with pytest.raises(ValidationError):
Settings(_env_file=None) # type: ignore[call-arg]
@pytest.mark.parametrize("length", [0, 1, 31])
def test_密钥短于32字符时拒绝(self, length: int) -> None:
with pytest.raises(ValidationError):
_settings(app_secret_key="x" * length)
def test_密钥恰好32字符时通过(self) -> None:
assert _settings(app_secret_key="x" * 32).app_secret_key == "x" * 32
class TestSecurityBounds:
"""安全边界:规范要求 JWT ≤120 分钟、验证码 300 秒。"""
def test_jwt有效期超过120分钟时拒绝(self) -> None:
with pytest.raises(ValidationError):
_settings(jwt_expire_minutes=121)
def test_jwt有效期为120分钟时通过(self) -> None:
assert _settings(jwt_expire_minutes=120).jwt_expire_minutes == 120
def test_验证码有效期超过300秒时拒绝(self) -> None:
with pytest.raises(ValidationError):
_settings(captcha_expire_seconds=301)
def test_验证码有效期为300秒时通过(self) -> None:
assert _settings(captcha_expire_seconds=300).captcha_expire_seconds == 300
class TestConnectionUrl:
"""连接 URL 构造:密码含特殊字符时不可手工拼接。"""
def test_数据库URL使用同步驱动(self) -> None:
"""运行期与迁移共用同步连接串,不存在同步/异步两套配置。"""
url = _settings().database_url
assert url.drivername == "mysql+pymysql"
assert url.database == "payroll"
assert url.query.get("charset") == "utf8mb4"
def test_密码含特殊字符时被原样保存于URL对象(self) -> None:
url = _settings(mysql_password="p@ss:w/ord#1").database_url
# URL.create 以结构化方式保存各组件,不做编码 —— 编码发生在驱动建连时,
# 这正是手工拼接字符串会出错(如把 : 误认为端口分隔符)而它不会的原因
assert url.password == "p@ss:w/ord#1"
assert url.username == "payroll"
assert url.port == 3306
def test_渲染URL时默认隐藏密码(self) -> None:
"""异常或日志中打印 URL 时不得泄露密码。"""
url = _settings(mysql_password="p@ss:w/ord#1").database_url
assert "p@ss:w/ord#1" not in url.render_as_string()
def test_Redis无密码时不拼接认证段(self) -> None:
assert _settings(redis_password="").redis_url == "redis://127.0.0.1:6379/0"
def test_Redis有密码时拼接认证段(self) -> None:
url = _settings(redis_password="secret").redis_url
assert url == "redis://:secret@127.0.0.1:6379/0"
@@ -0,0 +1,61 @@
"""全局异常处理器单元测试。
重点验证校验错误的文案压缩逻辑:FastAPI 默认返回 422 与 detail 数组,
本项目需映射为 400 与单行中文提示。
"""
from fastapi.exceptions import RequestValidationError
from app.core.constants import CODE_BAD_REQUEST
from app.core.exception_handlers import _build_response, _format_validation_errors
class TestFormatValidationErrors:
"""校验错误文案压缩。"""
def test_包含字段名与原始说明(self) -> None:
exc = RequestValidationError(
[{"loc": ("body", "username"), "msg": "Field required", "type": "missing"}]
)
assert _format_validation_errors(exc) == "参数错误:username Field required"
def test_去掉来源前缀只保留字段路径(self) -> None:
exc = RequestValidationError(
[
{
"loc": ("query", "pageNum"),
"msg": "Input should be greater than 0",
"type": "greater_than",
}
]
)
assert "query" not in _format_validation_errors(exc)
assert "pageNum" in _format_validation_errors(exc)
def test_多个错误时只取第一条(self) -> None:
exc = RequestValidationError(
[
{"loc": ("body", "a"), "msg": "错误甲", "type": "missing"},
{"loc": ("body", "b"), "msg": "错误乙", "type": "missing"},
]
)
message = _format_validation_errors(exc)
assert "错误甲" in message
assert "错误乙" not in message
def test_错误列表为空时兜底(self) -> None:
assert _format_validation_errors(RequestValidationError([])) == "参数错误"
class TestBuildResponse:
"""统一响应构造:HTTP 状态码与响应体 code 必须一致。"""
def test_状态码与响应体code一致(self) -> None:
response = _build_response(CODE_BAD_REQUEST, "参数错误")
assert response.status_code == CODE_BAD_REQUEST
import json
body = json.loads(response.body)
assert body["code"] == CODE_BAD_REQUEST
assert body["message"] == "参数错误"
assert body["data"] is None
+112
View File
@@ -0,0 +1,112 @@
"""JWT 签发与校验单元测试。
守的是:令牌的**有效期语义**与**版本号机制**。
这两点出错都不会在开发期暴露,但后果严重 ——
有效期算错 8 小时会表现为「刚登录就过期」或「过期了还能用」;
版本号机制失效则意味着禁用账号后旧令牌仍可访问全员薪酬。
"""
import uuid
from datetime import UTC, datetime, timedelta
import jwt
import pytest
from app.core.config import get_settings
from app.core.constants import JWT_ALGORITHM
from app.core.exceptions import AuthError
from app.core.security import create_access_token, decode_token
class TestCreateAccessToken:
"""令牌签发。"""
def test_返回令牌_jti与有效期(self) -> None:
token, jti, expires_in = create_access_token(user_id=1, token_version=1, roles=["admin"])
assert isinstance(token, str) and token.count(".") == 2 # header.payload.signature
assert len(jti) == 32
assert expires_in == get_settings().jwt_expire_minutes * 60
def test_载荷包含鉴定所需的全部字段(self) -> None:
token, jti, _ = create_access_token(user_id=7, token_version=3, roles=["admin", "user"])
payload = decode_token(token)
assert payload["sub"] == "7" # PyJWT 要求 sub 为字符串
assert payload["jti"] == jti
assert payload["ver"] == 3
assert payload["roles"] == ["admin", "user"]
def test_每次签发的jti都不同(self) -> None:
"""jti 是登出拉黑的唯一键,重复会导致误拉黑他人令牌。"""
jtis = set()
for _ in range(20):
_, jti, _ = create_access_token(user_id=1, token_version=1, roles=[])
jtis.add(jti)
assert len(jtis) == 20
def test_有效期基于UTC而非本地时间(self) -> None:
"""库列存本地时间、JWT 用 UTC,两者混算会产生 8 小时偏差。
这里断言 exp 与 UTC 当前时间的差值确实等于配置值 ——
若误用 datetime.now()(本地时间)生成,差值会差 8 小时。
"""
token, _, expires_in = create_access_token(user_id=1, token_version=1, roles=[])
exp = decode_token(token)["exp"]
drift = exp - datetime.now(UTC).timestamp()
assert abs(drift - expires_in) < 5 # 允许几秒执行耗时
class TestDecodeToken:
"""令牌解析与校验。"""
def test_正常令牌可解析(self) -> None:
token, _, _ = create_access_token(user_id=1, token_version=1, roles=["admin"])
assert decode_token(token)["sub"] == "1"
def test_已过期令牌抛AuthError(self) -> None:
"""直接构造过期令牌 —— create_access_token 的最小有效期为 1 分钟,
无法用它验证过期分支。"""
settings = get_settings()
past = datetime.now(UTC) - timedelta(hours=1)
expired = jwt.encode(
{
"sub": "1",
"jti": uuid.uuid4().hex,
"ver": 1,
"iat": past,
"exp": past + timedelta(seconds=60),
},
settings.app_secret_key,
algorithm=JWT_ALGORITHM,
)
with pytest.raises(AuthError):
decode_token(expired)
def test_签名被篡改时抛AuthError(self) -> None:
forged = jwt.encode(
{
"sub": "1",
"jti": uuid.uuid4().hex,
"ver": 1,
"exp": datetime.now(UTC) + timedelta(hours=1),
},
"wrong-secret-key-that-is-long-enough-to-pass-validation",
algorithm=JWT_ALGORITHM,
)
with pytest.raises(AuthError):
decode_token(forged)
@pytest.mark.parametrize("garbage", ["", "not.a.token", "a.b", "...."])
def test_格式非法的令牌抛AuthError而非其他异常(self, garbage: str) -> None:
"""令牌内容不可信,任何畸形输入都必须收敛为 401 而不是 500。"""
with pytest.raises(AuthError):
decode_token(garbage)
def test_算法被改为none时拒绝(self) -> None:
"""经典的 JWT 算法混淆攻击:把 alg 改成 none 伪造无签名令牌。"""
unsigned = jwt.encode(
{"sub": "1", "exp": datetime.now(UTC) + timedelta(hours=1)},
key="",
algorithm="none",
)
with pytest.raises(AuthError):
decode_token(unsigned)
+111
View File
@@ -0,0 +1,111 @@
"""日志脱敏单元测试。
规范第 8 节禁止日志输出密码、手机号、身份证、Token。
脱敏是最容易在重构中悄悄失效的一环,故用参数化用例覆盖各条规则。
"""
import logging
import pytest
from app.core.logging import SensitiveFormatter, scrub
class TestScrub:
"""文本脱敏规则。"""
@pytest.mark.parametrize(
("raw", "leaked"),
[
('{"password": "Admin@123"}', "Admin@123"),
("password=Secret123", "Secret123"),
("pwd: hunter2xyz", "hunter2xyz"),
('{"app_secret_key": "abcdef123456"}', "abcdef123456"),
("token=eyJhbGciOiJIUzI1NiJ9", "eyJhbGciOiJIUzI1NiJ9"),
],
)
def test_凭据值被脱敏(self, raw: str, leaked: str) -> None:
assert leaked not in scrub(raw)
def test_Bearer令牌被脱敏(self) -> None:
result = scrub("Authorization: Bearer eyJhbGciOi.J9.abc-_123")
assert "eyJhbGciOi.J9.abc-_123" not in result
assert "Bearer ***" in result
def test_手机号保留前三后四(self) -> None:
assert scrub("联系电话 13812345678") == "联系电话 138****5678"
def test_身份证保留前六后四(self) -> None:
assert scrub("证件号 110101199003071234") == "证件号 110101********1234"
def test_普通文本不被误伤(self) -> None:
text = "用户 admin 登录成功,耗时 120ms"
assert scrub(text) == text
def _format(msg: str, args: object = None) -> str:
"""用 SensitiveFormatter 渲染一条日志,返回最终文本。"""
record = logging.LogRecord(
name="t",
level=logging.INFO,
pathname="",
lineno=0,
msg=msg,
args=args,
exc_info=None,
)
return SensitiveFormatter("%(message)s").format(record)
class TestSensitiveFormatter:
"""脱敏必须在 %-格式化之后进行。"""
def test_脱敏位置参数中的敏感值(self) -> None:
# 回归用例:早期实现在 filter 阶段改写 record.msg,把 %s 占位符一并抹掉,
# 导致 logging 内部 msg % args 抛 TypeError,整条日志丢失
result = _format("登录失败 password=%s", ("Secret123",))
assert "Secret123" not in result
assert "登录失败" in result
def test_脱敏字典参数中的敏感值(self) -> None:
# LogRecord 要求 args 为元组;字典参数须包成单元素元组,
# 这正是 logger.info(msg, {...}) 的内部形态
result = _format("token=%(token)s", ({"token": "abc123xyz"},))
assert "abc123xyz" not in result
def test_脱敏消息模板内联的敏感值(self) -> None:
"""敏感值直接写在消息模板里(无占位符)时同样要脱敏。"""
result = _format('请求体 {"password": "Admin@123"}')
assert "Admin@123" not in result
def test_普通日志不被改动(self) -> None:
assert (
_format("用户 admin 登录成功,耗时 %d ms", (120,)) == "用户 admin 登录成功,耗时 120 ms"
)
def test_无参数日志正常通过(self) -> None:
assert _format("服务已启动") == "服务已启动"
class TestSetupLogging:
"""setup_logging 必须挂上脱敏格式化器,否则规则形同虚设。"""
def test_所有handler均使用脱敏格式化器(self) -> None:
import logging as _logging
from app.core.logging import setup_logging
setup_logging(level="INFO", log_to_file=False)
handlers = _logging.getLogger().handlers
assert handlers, "根日志器应至少有一个 handler"
assert all(isinstance(h.formatter, SensitiveFormatter) for h in handlers)
def test_重复调用不会累积handler(self) -> None:
import logging as _logging
from app.core.logging import setup_logging
setup_logging(log_to_file=False)
first = len(_logging.getLogger().handlers)
setup_logging(log_to_file=False)
assert len(_logging.getLogger().handlers) == first
+82
View File
@@ -0,0 +1,82 @@
"""登录失败限流单元测试。
守的是:锁定阈值与**滑动窗口**语义。若记录失败时不刷新 TTL,
「4 次失败 + 等 14 分钟 + 1 次失败」就不会触发锁定,
正好给慢速撞库留出缝隙。
"""
from app.core.config import get_settings
from app.core.constants import REDIS_KEY_LOGIN_FAIL
from app.services import login_limiter
def _ttl(username: str) -> int:
from app.core.redis_client import get_redis
return get_redis().ttl(REDIS_KEY_LOGIN_FAIL.format(username=username))
class TestThreshold:
"""阈值判定。"""
def test_初始状态未锁定且计数为零(self, fake_redis) -> None:
assert login_limiter.failure_count("u") == 0
assert login_limiter.is_locked("u") is False
def test_未达阈值时不锁定(self, fake_redis) -> None:
for _ in range(get_settings().login_fail_limit - 1):
login_limiter.record_failure("u")
assert login_limiter.is_locked("u") is False
def test_达到阈值即锁定(self, fake_redis) -> None:
for _ in range(get_settings().login_fail_limit):
login_limiter.record_failure("u")
assert login_limiter.is_locked("u") is True
def test_不同用户之间互不影响(self, fake_redis) -> None:
"""按用户名隔离,避免一个账号被攻击导致全员被锁。"""
for _ in range(get_settings().login_fail_limit):
login_limiter.record_failure("victim")
assert login_limiter.is_locked("victim") is True
assert login_limiter.is_locked("others") is False
class TestSlidingWindow:
"""滑动窗口:每次失败都刷新过期时间。"""
def test_锁定时长与配置一致(self, fake_redis) -> None:
"""容忍 1 秒偏差:Redis 的 TTL 向下取整,刚写入就查询可能少 1 秒。"""
login_limiter.record_failure("u")
expected = get_settings().login_lock_minutes * 60
assert _ttl("u") in (expected, expected - 1)
def test_再次失败会刷新过期时间(self, fake_redis) -> None:
"""把 TTL 改小模拟「距上次失败已过 14 分钟」,再失败一次应重新计满。
若不刷新,此处 TTL 仍是改小后的值,"慢速撞库"就能绕过锁定。
"""
login_limiter.record_failure("u")
fake_redis.expire(REDIS_KEY_LOGIN_FAIL.format(username="u"), 10)
login_limiter.record_failure("u")
assert _ttl("u") > 600
class TestClearAndRemaining:
"""解锁与剩余时长。"""
def test_登录成功清除失败计数(self, fake_redis) -> None:
for _ in range(get_settings().login_fail_limit):
login_limiter.record_failure("u")
login_limiter.clear("u")
assert login_limiter.is_locked("u") is False
assert login_limiter.failure_count("u") == 0
def test_未锁定时剩余秒数为零(self, fake_redis) -> None:
assert login_limiter.remaining_lock_seconds("u") == 0
def test_锁定后剩余秒数为正且不超过锁定时长(self, fake_redis) -> None:
for _ in range(get_settings().login_fail_limit):
login_limiter.record_failure("u")
remaining = login_limiter.remaining_lock_seconds("u")
assert 0 < remaining <= get_settings().login_lock_minutes * 60
+127
View File
@@ -0,0 +1,127 @@
"""ORM 模型元数据测试。
建表规范(表小写蛇形复数、字段中文 COMMENT、主键 id、索引 uk_/idx_ 前缀)
属于「红线」,靠人工 review 迟早会漏,这里用测试把它固定下来。
"""
import re
import pytest
from sqlalchemy.dialects import mysql
from app.models import Base
EXPECTED_TABLES = {"users", "roles", "user_roles", "permissions", "role_permissions", "login_logs"}
def _ddl(table_name: str) -> str:
"""返回某张表的建表 DDL 文本。"""
from sqlalchemy.schema import CreateTable
table = Base.metadata.tables[table_name]
return str(CreateTable(table).compile(dialect=mysql.dialect()))
class TestTableSet:
"""表集合必须完整 —— 漏一张就会让 alembic 生成「删除该表」的迁移。"""
def test_表集合与预期一致(self) -> None:
assert set(Base.metadata.tables) == EXPECTED_TABLES
def test_所有表名均为小写蛇形复数(self) -> None:
for name in Base.metadata.tables:
assert re.fullmatch(r"[a-z][a-z0-9_]*", name), f"表名 {name} 不符合小写蛇形"
assert name.endswith("s"), f"表名 {name} 应为复数"
class TestNamingConvention:
"""约束命名:规范要求 uk_表_字段 / idx_表_字段。"""
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
def test_主键命名为pk_表名(self, table_name: str) -> None:
pk = Base.metadata.tables[table_name].primary_key
assert pk.name == f"pk_{table_name}"
def test_唯一键以uk_开头(self) -> None:
for table in Base.metadata.tables.values():
for constraint in table.constraints:
if constraint.__class__.__name__ == "UniqueConstraint":
assert constraint.name.startswith(f"uk_{table.name}_"), constraint.name
def test_索引以idx_开头(self) -> None:
for table in Base.metadata.tables.values():
for index in table.indexes:
assert index.name.startswith(f"idx_{table.name}_"), index.name
def test_外键以fk_开头(self) -> None:
for table in Base.metadata.tables.values():
for fk in table.foreign_keys:
assert fk.constraint.name.startswith(f"fk_{table.name}_"), fk.constraint.name
class TestComments:
"""规范红线要求中文 COMMENT。"""
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
def test_表级中文注释存在(self, table_name: str) -> None:
assert Base.metadata.tables[table_name].comment, f"{table_name} 缺少表注释"
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
def test_每列都有中文注释(self, table_name: str) -> None:
table = Base.metadata.tables[table_name]
missing = [c.name for c in table.columns if not c.comment]
assert not missing, f"{table_name} 缺注释的列:{missing}"
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
def test_注释确实输出到DDL(self, table_name: str) -> None:
ddl = _ddl(table_name)
assert "COMMENT '" in ddl
assert f"COMMENT='{Base.metadata.tables[table_name].comment}'" in ddl
class TestMysqlTableOptions:
"""引擎与字符集:不显式指定则跟随服务端,跨环境可能不一致。"""
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
def test_显式指定引擎与字符集(self, table_name: str) -> None:
ddl = _ddl(table_name)
assert "ENGINE=InnoDB" in ddl
assert "CHARSET=utf8mb4" in ddl
assert "utf8mb4_general_ci" in ddl
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
def test_主键为无符号BIGINT(self, table_name: str) -> None:
assert "id BIGINT UNSIGNED NOT NULL" in _ddl(table_name)
class TestTimestampColumns:
"""更新时间必须由数据库维护 —— 这是 server_onupdate 做不到的。"""
@pytest.mark.parametrize("table_name", ["users", "roles", "permissions"])
def test_带时间戳的表含ON_UPDATE子句(self, table_name: str) -> None:
assert "ON UPDATE CURRENT_TIMESTAMP" in _ddl(table_name)
@pytest.mark.parametrize("table_name", ["login_logs"])
def test_仅追加的日志表不含ON_UPDATE子句(self, table_name: str) -> None:
"""日志是不可变记录,没有"更新时间"语义。"""
assert "ON UPDATE CURRENT_TIMESTAMP" not in _ddl(table_name)
def test_关联表保留create_time便于审计(self) -> None:
for table_name in ("user_roles", "role_permissions"):
assert "create_time" in Base.metadata.tables[table_name].columns
class TestEnumColumns:
"""状态类字段的默认值。"""
def test_用户默认启用且令牌版本为1(self) -> None:
users = Base.metadata.tables["users"]
assert users.columns["status"].server_default.arg == "1"
assert users.columns["token_version"].server_default.arg == "1"
def test_角色编码唯一(self) -> None:
ddl = _ddl("roles")
assert "uk_roles_role_code UNIQUE (role_code)" in ddl
def test_用户名唯一(self) -> None:
assert "uk_users_username UNIQUE (username)" in _ddl("users")
+67
View File
@@ -0,0 +1,67 @@
"""统一响应模型单元测试。"""
import pytest
from pydantic import ValidationError
from app.core.response import ApiResponse, PageQuery, PageResult, ok
class TestApiResponse:
"""响应信封结构。"""
def test_默认值符合规范(self) -> None:
resp = ApiResponse[None]()
assert resp.code == 200
assert resp.message == "success"
assert resp.data is None
def test_序列化字段固定为code_message_data(self) -> None:
keys = set(ApiResponse[None]().model_dump().keys())
assert keys == {"code", "message", "data"}
class TestOkHelper:
"""成功响应构造助手。"""
def test_携带业务数据(self) -> None:
resp = ok({"id": 1})
assert resp.code == 200
assert resp.data == {"id": 1}
def test_可自定义提示语(self) -> None:
assert ok(message="登录成功").message == "登录成功"
class TestPageResult:
"""分页结果:Python 蛇形属性名,序列化为规范要求的驼峰字段。"""
def test_序列化输出驼峰字段(self) -> None:
dumped = PageResult[int](records=[1, 2], total=2, page_num=1, page_size=10).model_dump(
by_alias=True
)
assert dumped == {"list": [1, 2], "total": 2, "pageNum": 1, "pageSize": 10}
def test_空结果时list为空数组而非null(self) -> None:
dumped = PageResult[int]().model_dump(by_alias=True)
assert dumped["list"] == []
class TestPageQuery:
"""分页入参。"""
def test_offset按页码换算(self) -> None:
assert PageQuery(page_num=1, page_size=10).offset == 0
assert PageQuery(page_num=3, page_size=20).offset == 40
@pytest.mark.parametrize("bad_page", [0, -1])
def test_页码小于1时拒绝(self, bad_page: int) -> None:
with pytest.raises(ValidationError):
PageQuery(page_num=bad_page)
def test_每页条数超上限时拒绝(self) -> None:
with pytest.raises(ValidationError):
PageQuery(page_size=201)
def test_接受驼峰输入(self) -> None:
query = PageQuery.model_validate({"pageNum": 2, "pageSize": 5})
assert (query.page_num, query.page_size) == (2, 5)
+66
View File
@@ -0,0 +1,66 @@
"""密码哈希单元测试。
重点覆盖 bcrypt 5.0 的 72 字节硬限制 —— 此前是静默截断,现在是抛 ValueError,
若不在边界上把关,用户设一个长密码就会在登录时炸出 500。
"""
import pytest
from app.core.constants import PASSWORD_MAX_BYTES
from app.core.security import hash_password, verify_password
class TestHashPassword:
"""哈希生成。"""
def test_输出为BCrypt格式且定长60字符(self) -> None:
hashed = hash_password("Admin@2026")
assert hashed.startswith("$2")
assert len(hashed) == 60
def test_相同明文两次哈希结果不同(self) -> None:
"""BCrypt 内含随机盐,相同明文每次结果不同 —— 这是防彩虹表的基础。"""
assert hash_password("SamePwd@1") != hash_password("SamePwd@1")
def test_恰好72字节时允许(self) -> None:
password = "a" * PASSWORD_MAX_BYTES
assert len(password.encode("utf-8")) == PASSWORD_MAX_BYTES
assert len(hash_password(password)) == 60
def test_超过72字节时抛ValueError(self) -> None:
with pytest.raises(ValueError):
hash_password("a" * (PASSWORD_MAX_BYTES + 1))
@pytest.mark.parametrize(
("text", "count", "should_pass"),
[("中", 24, True), ("中", 25, False)],
)
def test_中文按字节而非字符计算边界(self, text: str, count: int, should_pass: bool) -> None:
"""一个汉字 3 字节:24 个正好 72 字节,25 个即 75 字节超限。"""
password = text * count
if should_pass:
assert len(hash_password(password)) == 60
else:
with pytest.raises(ValueError):
hash_password(password)
class TestVerifyPassword:
"""哈希校验。"""
def test_正确密码校验通过(self) -> None:
assert verify_password("Admin@2026", hash_password("Admin@2026")) is True
def test_错误密码校验失败(self) -> None:
assert verify_password("WrongPwd@1", hash_password("Admin@2026")) is False
def test_超长密码返回False而不抛异常(self) -> None:
"""校验路径不得因输入长度报错 —— 否则会形成「密码过长」与「密码错误」
可区分的侧信道,且会让恶意长输入直接打出 500。"""
assert verify_password("a" * 200, hash_password("Admin@2026")) is False
def test_损坏的哈希串返回False而不抛异常(self) -> None:
assert verify_password("Admin@2026", "not-a-bcrypt-hash") is False
def test_空哈希串返回False(self) -> None:
assert verify_password("Admin@2026", "") is False
@@ -0,0 +1,60 @@
"""登出黑名单单元测试。
守的是:黑名单的 **TTL 必须等于令牌剩余有效期**。
写死固定值会出现「登出 N 分钟后令牌又活了」;
完全不设过期则 Redis 会随登出次数无限增长。两者都是真实事故。
"""
from app.core.constants import REDIS_KEY_TOKEN_BLACKLIST
from app.services import token_blacklist
def _ttl(jti: str) -> int:
from app.core.redis_client import get_redis
return get_redis().ttl(REDIS_KEY_TOKEN_BLACKLIST.format(jti=jti))
class TestRevoke:
"""拉黑。"""
def test_拉黑后可检出(self, fake_redis) -> None:
token_blacklist.revoke("jti-1", 600)
assert token_blacklist.is_revoked("jti-1") is True
def test_TTL等于传入的剩余秒数(self, fake_redis) -> None:
"""容忍 1 秒偏差:Redis 的 TTL 以毫秒精度存绝对过期时刻,
命令返回时向下取整,刚写入就查询必然是 N 或 N-1。
断言死等会变成偶发失败的"闪烁测试"。"""
token_blacklist.revoke("jti-1", 1234)
assert _ttl("jti-1") in (1234, 1233)
def test_键带过期时间而非永久存在(self, fake_redis) -> None:
"""永久存储会让 Redis 随登出次数无限增长。"""
token_blacklist.revoke("jti-1", 600)
assert _ttl("jti-1") > 0
def test_已过期令牌不写入(self, fake_redis) -> None:
"""剩余有效期为 0 或负数时无凭证可撤销,写入只会浪费内存。"""
token_blacklist.revoke("jti-expired", 0)
token_blacklist.revoke("jti-expired2", -100)
assert token_blacklist.is_revoked("jti-expired") is False
assert token_blacklist.is_revoked("jti-expired2") is False
def test_重复拉黑同一jti不报错且TTL更新(self, fake_redis) -> None:
token_blacklist.revoke("jti-1", 600)
token_blacklist.revoke("jti-1", 300)
assert _ttl("jti-1") in (300, 299)
class TestIsRevoked:
"""检测。"""
def test_未拉黑的令牌返回False(self, fake_redis) -> None:
assert token_blacklist.is_revoked("never-revoked") is False
def test_不同jti之间互不影响(self, fake_redis) -> None:
"""jti 是唯一键,串号会导致误踢其他用户的登录态。"""
token_blacklist.revoke("jti-1", 600)
assert token_blacklist.is_revoked("jti-1") is True
assert token_blacklist.is_revoked("jti-2") is False