小蚂蚁云

Appearance

Become a sponsor
企业级中台解决方案
企业级中台解决方案

测试规范 ​

概述 ​

项目采用三层测试体系：契约快照测试、单元测试、集成测试。测试使用 pytest 框架，分层覆盖不同维度的质量保障。

三层测试体系

  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 的集成测试，默认跳过

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_*.pytest_*
单元测试test_*.pytest_*
集成测试test_*.pytest_*

函数命名规则

测试函数名应清晰描述被测行为：

  • 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

集成测试环境

集成测试需要真实的 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

覆盖率目标

  • Repository 层：80%+
  • Service 层：70%+
  • Endpoint 层：由契约测试保障路由正确性

总结 ​

测试规范采用三层体系：契约快照测试保障路由/权限不变、单元测试验证单函数逻辑、集成测试验证模块协作。测试文件命名 test_*.py，函数命名 test_*。pytest 标记区分测试类型，默认跳过集成测试。契约快照通过 UPDATE_GOLDEN=1 环境变量更新基线。

小蚂蚁云团队 · 提供技术支持

Copyright © 2020~2026 www.djangoadmin.cn. All rights reserved.