405 lines
12 KiB
Markdown
405 lines
12 KiB
Markdown
# 测试规范
|
||
|
||
## 概述
|
||
|
||
项目采用三层测试体系:契约快照测试、单元测试、集成测试。测试使用 pytest 框架,分层覆盖不同维度的质量保障。
|
||
|
||
:::tip 三层测试体系
|
||
1. 契约测试(Contract):验证路由、权限、表名等契约不变
|
||
2. 单元测试(Unit):验证单个函数/方法的逻辑正确性
|
||
3. 集成测试(Integration):验证模块间协作、真实数据库操作
|
||
:::
|
||
|
||
## 目录结构
|
||
|
||
```
|
||
tests/
|
||
├── contract/ # 契约快照测试
|
||
│ ├── golden/ # 快照基线文件(JSON)
|
||
│ │ ├── routes.json
|
||
│ │ ├── permissions.json
|
||
│ │ └── tables.json
|
||
│ ├── helpers.py # 测试辅助函数
|
||
│ ├── test_routes_snapshot.py # 路由路径快照
|
||
│ ├── test_permission_snapshot.py # 权限标识快照
|
||
│ ├── test_tablename_snapshot.py # 表名快照
|
||
│ └── test_schema_sql_consistency.py # 模型与 SQL 一致性
|
||
├── unit/ # 单元测试
|
||
│ ├── core/ # 核心模块测试
|
||
│ │ ├── test_jwt.py
|
||
│ │ ├── test_pagination.py
|
||
│ │ ├── test_response.py
|
||
│ │ └── ...
|
||
│ ├── middleware/ # 中间件测试
|
||
│ │ ├── test_operation_log.py
|
||
│ │ └── ...
|
||
│ ├── modules/ # 业务模块测试
|
||
│ │ ├── conftest.py # service 层公共 fixture
|
||
│ │ ├── test_upload_service.py
|
||
│ │ ├── test_base_service_page.py
|
||
│ │ └── ...
|
||
│ ├── utils/ # 工具函数测试
|
||
│ │ ├── test_ip_trusted_proxy.py
|
||
│ │ ├── test_ip2region.py
|
||
│ │ └── ...
|
||
│ └── scripts/ # 脚本测试
|
||
│ └── test_migrate_db.py
|
||
├── integration/ # 集成测试(默认跳过)
|
||
│ └── api/
|
||
│ ├── test_smoke.py
|
||
│ └── test_transaction.py
|
||
├── test_bugfixes.py # Bug 回归测试
|
||
├── test_rich_text.py # 富文本工具测试
|
||
└── test_middleware_smoke.py # 中间件冒烟测试
|
||
```
|
||
|
||
## pytest 配置
|
||
|
||
```ini
|
||
# pytest.ini
|
||
[pytest]
|
||
minversion = 7.0
|
||
addopts = -ra --strict-markers -m "not integration"
|
||
testpaths = tests
|
||
pythonpath = src
|
||
|
||
markers =
|
||
contract: 契约快照测试(路由/权限串/表名),守护前端与数据契约
|
||
integration: 需要真实 MySQL/Redis 的集成测试,默认跳过
|
||
```
|
||
|
||
:::tip pythonpath
|
||
`pythonpath = src` 让测试文件可以直接 `from modules.xxx import ...` 和 `from core.xxx import ...`,无需手动设置 PYTHONPATH。
|
||
:::
|
||
|
||
## 契约快照测试
|
||
|
||
验证路由路径、权限标识、数据库表名等契约不变。快照文件存储在 `tests/contract/golden/` 目录。
|
||
|
||
### 路由快照
|
||
|
||
```python
|
||
# tests/contract/test_routes_snapshot.py
|
||
"""路由契约快照:守护全部 API 路径与 HTTP 方法不变。"""
|
||
import pytest
|
||
from tests.contract.helpers import collect_routes, compare_snapshot
|
||
|
||
pytestmark = pytest.mark.contract
|
||
|
||
|
||
def test_routes_snapshot():
|
||
compare_snapshot("routes.json", collect_routes())
|
||
```
|
||
|
||
### 权限快照
|
||
|
||
```python
|
||
# tests/contract/test_permission_snapshot.py
|
||
"""权限契约快照:守护全部 @permission_required 权限串不变。"""
|
||
import pytest
|
||
from tests.contract.helpers import collect_permissions, compare_snapshot
|
||
|
||
pytestmark = pytest.mark.contract
|
||
|
||
|
||
def test_permissions_snapshot():
|
||
compare_snapshot("permissions.json", collect_permissions())
|
||
```
|
||
|
||
### 辅助函数
|
||
|
||
```python
|
||
# tests/contract/helpers.py
|
||
import json
|
||
import os
|
||
|
||
GOLDEN_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "golden")
|
||
UPDATE_GOLDEN = os.getenv("UPDATE_GOLDEN") == "1"
|
||
|
||
|
||
def collect_routes():
|
||
"""通过 openapi schema 收集全部 (METHOD /path),去重排序。"""
|
||
from core.app import create_app
|
||
app = create_app()
|
||
spec = app.openapi()
|
||
return sorted(
|
||
f"{method.upper()} {path}"
|
||
for path, methods in spec["paths"].items()
|
||
for method in methods
|
||
)
|
||
|
||
|
||
def collect_permissions():
|
||
"""静态扫描源码中的 @permission_required("sys:...") 权限串。"""
|
||
import glob
|
||
import re
|
||
perms = set()
|
||
PERMISSION_RE = re.compile(r'@permission_required\(\s*["\'](sys:[^"\']+)["\']')
|
||
for root in ("src/api/v1/endpoints", "src/modules"):
|
||
for path in glob.glob(os.path.join(root, "**", "*.py"), recursive=True):
|
||
with open(path, encoding="utf-8") as f:
|
||
perms.update(PERMISSION_RE.findall(f.read()))
|
||
return sorted(perms)
|
||
|
||
|
||
def compare_snapshot(name: str, actual: list):
|
||
"""与 golden 对比;UPDATE_GOLDEN=1 时重新生成并跳过断言。"""
|
||
path = os.path.join(GOLDEN_DIR, name)
|
||
if UPDATE_GOLDEN:
|
||
os.makedirs(GOLDEN_DIR, exist_ok=True)
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(actual, f, ensure_ascii=False, indent=2)
|
||
return
|
||
with open(path, encoding="utf-8") as f:
|
||
golden = json.load(f)
|
||
assert actual == golden, (
|
||
f"契约快照 [{name}] 不一致!\n"
|
||
f"新增: {sorted(set(actual) - set(golden))}\n"
|
||
f"缺失: {sorted(set(golden) - set(actual))}"
|
||
)
|
||
```
|
||
|
||
### 更新快照
|
||
|
||
```bash
|
||
# 更新全部快照基线
|
||
UPDATE_GOLDEN=1 pytest tests/contract -m contract
|
||
|
||
# 更新后运行一次验证
|
||
pytest tests/contract -m contract
|
||
```
|
||
|
||
## 单元测试
|
||
|
||
验证单个函数/方法的逻辑正确性,不依赖外部服务。
|
||
|
||
### 测试命名规范
|
||
|
||
| 类型 | 文件命名 | 函数命名 |
|
||
|------|---------|---------|
|
||
| 契约测试 | `test_*.py` | `test_*` |
|
||
| 单元测试 | `test_*.py` | `test_*` |
|
||
| 集成测试 | `test_*.py` | `test_*` |
|
||
|
||
:::tip 函数命名规则
|
||
测试函数名应清晰描述被测行为:
|
||
- `test_upload_file_rejects_svg`:验证上传文件拒绝 SVG
|
||
- `test_is_private_ip_handles_malformed_ip`:验证畸形 IP 不抛异常
|
||
- `test_batch_delete_soft_deletes`:验证批量删除执行软删除
|
||
:::
|
||
|
||
### monkeypatch 隔离依赖
|
||
|
||
使用 `monkeypatch` 替换模块级变量,隔离测试环境:
|
||
|
||
```python
|
||
# tests/unit/modules/test_upload_service.py
|
||
import asyncio
|
||
import json
|
||
from modules.upload import service as upload_service
|
||
from modules.upload.service import _save_upload, upload_file
|
||
|
||
|
||
class _FakeUpload:
|
||
"""模拟异步分块读取的 UploadFile。"""
|
||
def __init__(self, filename, data=b''):
|
||
self.filename = filename
|
||
self._data = data
|
||
|
||
async def read(self, size=65536):
|
||
data = self._data
|
||
self._data = b''
|
||
return data
|
||
|
||
|
||
def test_upload_file_rejects_svg(monkeypatch):
|
||
"""SVG 被 BANNED_UPLOAD_EXTS 硬拒绝"""
|
||
monkeypatch.setattr(upload_service, 'UPLOAD_ALLOWED_EXTS', '.svg')
|
||
result = asyncio.run(upload_file(_FakeUpload('logo.svg', b'<svg/>')))
|
||
body = json.loads(result.body)
|
||
assert body['ok'] is False
|
||
|
||
|
||
def test_save_upload_rejects_svg_even_if_config_allows():
|
||
"""兜底硬拒绝:即使配置放行 .svg,落盘前一律拒绝"""
|
||
content = b'<svg xmlns="http://www.w3.org/2000/svg"><script>alert(1)</script></svg>'
|
||
result = _save_upload(content, 'a.svg', '.svg')
|
||
body = json.loads(result.body)
|
||
assert body['ok'] is False
|
||
assert 'SVG' in body['msg']
|
||
```
|
||
|
||
### 畸形输入防御测试
|
||
|
||
```python
|
||
# tests/unit/utils/test_ip2region.py
|
||
from utils.ip2region import _is_private_ip, get_ip_location, get_ip_region
|
||
|
||
|
||
def test_is_private_ip_handles_malformed_ip():
|
||
"""畸形 IP 不再抛 ValueError,按非内网处理"""
|
||
assert _is_private_ip('172.abc') is False
|
||
assert _is_private_ip('172.') is False
|
||
|
||
|
||
def test_get_ip_location_malformed_ip_degrades_to_unknown():
|
||
"""畸形 IP 不抛异常,降级返回"未知" """
|
||
assert get_ip_location('172.abc') == '未知'
|
||
|
||
|
||
def test_is_private_ip_private_ranges():
|
||
assert _is_private_ip('172.16.5.5') is True
|
||
assert _is_private_ip('172.31.5.5') is True
|
||
assert _is_private_ip('172.32.5.5') is False
|
||
assert _is_private_ip('10.1.2.3') is True
|
||
assert _is_private_ip('192.168.1.1') is True
|
||
assert _is_private_ip('8.8.8.8') is False
|
||
```
|
||
|
||
### 代理白名单测试
|
||
|
||
```python
|
||
# tests/unit/utils/test_ip_trusted_proxy.py
|
||
import types
|
||
from utils import ip as ip_mod
|
||
from utils.ip import get_client_ip
|
||
|
||
|
||
def _make_request(client_host, xff=None):
|
||
"""构造模拟 Request 对象。"""
|
||
req = types.SimpleNamespace()
|
||
req.client = types.SimpleNamespace(host=client_host)
|
||
req.headers = {'X-Forwarded-For': xff} if xff else {}
|
||
return req
|
||
|
||
|
||
def test_proxy_count_zero_ignores_xff(monkeypatch):
|
||
"""TRUSTED_PROXY_COUNT=0:不信任代理头"""
|
||
monkeypatch.setattr(ip_mod, 'TRUSTED_PROXY_COUNT', 0)
|
||
monkeypatch.setattr(ip_mod, 'TRUSTED_PROXY_IPS', frozenset({'127.0.0.1'}))
|
||
assert get_client_ip(_make_request('203.0.113.9', xff='1.2.3.4')) == '203.0.113.9'
|
||
|
||
|
||
def test_peer_not_in_whitelist_rejects_xff(monkeypatch):
|
||
"""直连对端不在白名单 → XFF 被忽略"""
|
||
monkeypatch.setattr(ip_mod, 'TRUSTED_PROXY_COUNT', 1)
|
||
monkeypatch.setattr(ip_mod, 'TRUSTED_PROXY_IPS', frozenset({'10.0.0.1'}))
|
||
assert get_client_ip(_make_request('203.0.113.9', xff='1.2.3.4')) == '203.0.113.9'
|
||
|
||
|
||
def test_peer_in_whitelist_trusts_xff(monkeypatch):
|
||
"""单层代理:对端在白名单 → 信任 XFF"""
|
||
monkeypatch.setattr(ip_mod, 'TRUSTED_PROXY_COUNT', 1)
|
||
monkeypatch.setattr(ip_mod, 'TRUSTED_PROXY_IPS', frozenset({'10.0.0.1'}))
|
||
assert get_client_ip(_make_request('10.0.0.1', xff='1.2.3.4')) == '1.2.3.4'
|
||
|
||
|
||
def test_peer_in_whitelist_spoofed_left_entry_ignored(monkeypatch):
|
||
"""客户端伪造条目只会在 XFF 左侧"""
|
||
monkeypatch.setattr(ip_mod, 'TRUSTED_PROXY_COUNT', 1)
|
||
monkeypatch.setattr(ip_mod, 'TRUSTED_PROXY_IPS', frozenset({'10.0.0.1'}))
|
||
assert get_client_ip(_make_request('10.0.0.1', xff='6.6.6.6, 1.2.3.4')) == '1.2.3.4'
|
||
```
|
||
|
||
### service 层公共 fixture
|
||
|
||
```python
|
||
# tests/unit/modules/conftest.py
|
||
"""service 层单元测试公共 fixture:构造最小 Request 对象。"""
|
||
import types
|
||
import pytest
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_request():
|
||
"""模拟 FastAPI Request,仅提供 service 层用到的字段。"""
|
||
class FR:
|
||
state = types.SimpleNamespace(user_id=1, username="tester", realname="测试员")
|
||
query_params = {}
|
||
return FR()
|
||
```
|
||
|
||
## 集成测试
|
||
|
||
验证模块间协作和真实数据库操作(需真实 MySQL/Redis,默认跳过):
|
||
|
||
```python
|
||
import pytest
|
||
|
||
@pytest.mark.integration
|
||
class TestPositionIntegration:
|
||
"""岗位模块集成测试"""
|
||
|
||
def test_crud_flow(self, client, auth_headers):
|
||
"""验证完整的增删改查流程"""
|
||
# 添加
|
||
resp = client.post('/api/v1/position/add', json={
|
||
'name': '集成测试岗位', 'status': 1, 'sort': 1
|
||
}, headers=auth_headers)
|
||
assert resp.json()['ok'] is True
|
||
|
||
# 查询
|
||
resp = client.get('/api/v1/position/page?pageNo=1&pageSize=10', headers=auth_headers)
|
||
assert resp.json()['ok'] is True
|
||
assert resp.json()['data']['total'] > 0
|
||
```
|
||
|
||
:::warning 集成测试环境
|
||
集成测试需要真实的 MySQL 和 Redis 实例,默认跳过。运行前确保 `.env` 配置正确,且数据库中已有基础数据。
|
||
:::
|
||
|
||
## 运行命令
|
||
|
||
```bash
|
||
# 全部测试(跳过集成测试)
|
||
pytest
|
||
make test
|
||
|
||
# 契约测试
|
||
pytest tests/contract -m contract
|
||
make test-contract
|
||
|
||
# 单元测试
|
||
pytest tests/unit
|
||
make test-unit
|
||
|
||
# 集成测试(需真实 MySQL/Redis)
|
||
pytest tests/integration -m integration
|
||
make test-integration
|
||
|
||
# 运行指定文件
|
||
pytest tests/unit/modules/test_upload_service.py
|
||
|
||
# 运行指定函数
|
||
pytest tests/unit/modules/test_upload_service.py::test_upload_file_rejects_svg
|
||
|
||
# 显示详细输出
|
||
pytest -v
|
||
|
||
# 显示 print 输出
|
||
pytest -s
|
||
```
|
||
|
||
## 测试覆盖率
|
||
|
||
```bash
|
||
# 安装覆盖率插件
|
||
pip install pytest-cov
|
||
|
||
# 运行测试并生成覆盖率报告
|
||
pytest --cov=src --cov-report=html
|
||
|
||
# 查看报告
|
||
open htmlcov/index.html
|
||
```
|
||
|
||
:::tip 覆盖率目标
|
||
- Repository 层:80%+
|
||
- Service 层:70%+
|
||
- Endpoint 层:由契约测试保障路由正确性
|
||
:::
|
||
|
||
## 总结
|
||
|
||
测试规范采用三层体系:契约快照测试保障路由/权限不变、单元测试验证单函数逻辑、集成测试验证模块协作。测试文件命名 `test_*.py`,函数命名 `test_*`。pytest 标记区分测试类型,默认跳过集成测试。契约快照通过 `UPDATE_GOLDEN=1` 环境变量更新基线。
|