| name | yweb-testing |
| description | YWeb 框架 Pytest 单元测试编写规范。适用于 yweb-core 底层框架本身，以及通过 pip install yweb 安装后基于该框架开发的任何上层应用。涵盖测试目录结构、conftest 编写、fixture 模式、ORM 测试、API 测试、Service 测试、Domain 测试、Mock 策略、异步测试等场景。 |
YWeb Pytest 单元测试编写规范
本 Skill 是 AI 编程助手为 yweb 框架生态编写 pytest 测试文件的权威指南。适用于两种场景：
- yweb-core 底层框架本身：测试 ORM、缓存、认证、异常处理等基础模块
- 基于 yweb 开发的上层应用：通过
pip install yweb 安装框架后，测试应用自身的 API 层、Service 层、Domain 层业务逻辑
关于上层应用：本文档中的"上层应用"指任何通过 pip install yweb 引入框架后独立开发的项目。文档中以虚构的示例项目 my-app 演示目录结构和代码模板，实际项目名称和结构请根据自身项目调整。上层应用看不到 yweb-core 的源码和测试文件，只能通过 from yweb import ... 使用框架提供的公共 API。
核心原则：测试代码的质量标准与生产代码一致。测试应快速、可靠、可维护、易读。
1. 测试目录结构
1.1 yweb-core 底层框架
yweb-core/
├── pytest.ini # pytest 配置
├── tests/
│ ├── __init__.py
│ ├── conftest.py # 全局公共 fixtures
│ ├── helpers/ # 测试辅助工具（非测试类）
│ │ ├── __init__.py
│ │ ├── auth_helpers.py
│ │ └── transaction_helpers.py
│ ├── test_orm/
│ │ ├── __init__.py
│ │ ├── unit/ # 单元测试
│ │ │ ├── __init__.py
│ │ │ ├── test_base_model.py
│ │ │ └── test_query.py
│ │ └── integration/ # 集成测试
│ │ ├── __init__.py
│ │ └── test_transaction.py
│ ├── test_auth/
│ ├── test_cache/
│ ├── test_exceptions/
│ └── test_response/
1.2 上层应用（通过 pip install yweb 引入框架的独立项目）
上层应用是独立的项目，通过 pip install yweb 安装框架，只能使用 from yweb import ... 的公共 API。
应用看不到 yweb-core 的源码或测试文件，测试目录结构需要自行建立。
my-app/ # 你的项目根目录（名称自定义）
├── pytest.ini # pytest 配置（需创建）
├── tests/
│ ├── __init__.py
│ ├── conftest.py # 全局公共 fixtures（数据库、TestClient、认证等）
│ ├── helpers/ # 测试辅助工具
│ │ ├── __init__.py
│ │ ├── factory.py # 测试数据工厂
│ │ └── auth_utils.py # 认证辅助
│ ├── test_api/ # API 层测试（HTTP 端点）
│ │ ├── __init__.py
│ │ ├── test_role_api.py
│ │ └── test_user_api.py
│ ├── test_services/ # Service 层测试（业务编排）
│ │ ├── __init__.py
│ │ └── test_auth_service.py
│ ├── test_domain/ # Domain 层测试（领域模型 + 领域服务）
│ │ ├── __init__.py
│ │ ├── test_user_model.py
│ │ └── test_role_service.py
│ └── test_orm/ # ORM 功能测试
│ ├── __init__.py
│ └── test_models.py
1.3 目录命名规则
- 测试目录以
test_ 前缀命名，对应被测模块
- 大型模块拆分
unit/ 和 integration/ 子目录
helpers/ 目录放测试辅助工具（不是测试文件）- 每个测试目录都要有
__init__.py
2. pytest 配置
2.1 pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = -v --tb=short
markers =
slow: 标记为慢速测试
integration: 标记为集成测试
unit: 标记为单元测试
filterwarnings =
ignore::DeprecationWarning
ignore::PendingDeprecationWarning
2.2 常用依赖
# requirements-dev.txt
pytest>=8.0
pytest-asyncio>=0.23
pytest-mock>=3.12
pytest-cov>=4.1
httpx>=0.27 # 异步 TestClient 支持
3. 命名规范
3.1 测试文件命名
test_<模块名>.py # 例：test_base_model.py、test_role_api.py
3.2 测试类命名
class TestBaseModel:
class TestRoleService:
class TestRoleAPI:
3.3 测试方法命名
def test_create_user_with_valid_email(self):
def test_delete_role_raises_error_for_system_role(self):
def test_get_user_returns_none_when_not_found(self):
3.4 非测试类命名（关键！必须遵守！）
严重警告：pytest 的测试发现机制会自动将 test_*.py 文件中所有以 Test 开头的 class 识别为测试类，并尝试收集其中以 test_ 开头的方法作为测试用例。如果辅助类、实体类、模型类以 Test 开头，pytest 会产生警告或错误。
规则：只有真正的测试类才以 Test 开头，其他所有类一律禁止。
这条规则适用于测试文件中定义的所有非测试类，包括但不限于：
- 测试用 ORM 模型（如用于建表的 Model 类）
- Mock 类 / Stub 类
- 数据传输对象 / 辅助实体类
- 工厂类
- 配置类
class UserModel(BaseModel):
class MockUser:
class SampleRole:
class RoleFactory:
class FakeEmailService:
class TestUserModel(BaseModel):
class TestHelper:
class TestData:
记忆口诀：文件名 test_ 开头，测试类 Test 开头，测试方法 test_ 开头——只有这三者可以用 test 前缀，其余一切类和函数都不能以 test/Test 开头。
3.5 文档字符串（中文）
class TestBaseModel:
"""BaseModel 基础功能测试"""
def test_model_has_id(self):
"""测试模型创建后自动分配 ID"""
...
def test_soft_delete_restore(self):
"""测试软删除后恢复记录"""
...
4. conftest.py 编写规范
4.1 yweb-core 根 conftest.py 模板
"""
Pytest 公共配置和 Fixtures
提供测试所需的公共资源：
- 数据库连接（内存 SQLite）
- FastAPI 测试客户端
- JWT 相关 fixtures
- Mock 用户
"""
import pytest
import os
import tempfile
from typing import Generator
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session, scoped_session
from sqlalchemy.pool import StaticPool
from yweb.orm import CoreModel, BaseModel
def pytest_configure(config):
"""全局初始化钩子（最早执行，仅一次）"""
from yweb.orm import init_versioning, CurrentUserPlugin, is_versioning_initialized
if is_versioning_initialized():
return
enable_user_tracking = os.getenv('YWEB_TEST_USER_TRACKING', 'true').lower() == 'true'
if enable_user_tracking:
init_versioning(plugins=[CurrentUserPlugin()])
else:
init_versioning()
@pytest.fixture(scope="function")
def memory_engine():
"""创建内存数据库引擎（每个测试函数独立）
使用 StaticPool + check_same_thread=False 确保：
1. 所有操作使用同一个连接
2. 允许跨线程访问
"""
engine = create_engine(
"sqlite:///:memory:",
echo=False,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
yield engine
engine.dispose()
@pytest.fixture
def db_session(memory_engine) -> Generator[Session, None, None]:
"""创建数据库会话"""
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=memory_engine)
session = SessionLocal()
try:
yield session
finally:
session.close()
@pytest.fixture
def app():
"""创建测试用 FastAPI 应用"""
test_app = FastAPI(title="Test App")
return test_app
@pytest.fixture
def client(app):
"""创建测试客户端"""
return TestClient(app)
@pytest.fixture(scope="session")
def temp_dir():
"""创建临时目录（session 级别）"""
with tempfile.TemporaryDirectory() as tmpdir:
yield tmpdir
@pytest.fixture
def temp_file(temp_dir):
"""临时文件工厂"""
created_files = []
def _create_file(filename: str, content: str = "") -> str:
filepath = os.path.join(temp_dir, filename)
os.makedirs(os.path.dirname(filepath), exist_ok=True) if os.path.dirname(filepath) else None
with open(filepath, "w", encoding="utf-8") as f:
f.write(content)
created_files.append(filepath)
return filepath
yield _create_file
for f in created_files:
if os.path.exists(f):
os.remove(f)
4.2 上层应用 conftest.py 模板（适用于任何基于 yweb 的独立项目）
"""
应用层 Pytest 公共配置和 Fixtures
提供：
- 完整的应用实例（含路由、中间件）
- 已认证的测试客户端
- 测试数据库会话（自动清理）
- 测试用户和角色
注意：上层应用通过 pip install yweb 安装框架，
只能使用 from yweb import ... 的公共 API。
以下 import 路径中的 app.xxx 需要替换为你自己项目的实际模块路径。
"""
import pytest
from typing import Generator
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import StaticPool
from yweb.orm import CoreModel, BaseModel
@pytest.fixture(scope="function")
def memory_engine():
"""内存数据库引擎"""
engine = create_engine(
"sqlite:///:memory:",
echo=False,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
yield engine
engine.dispose()
@pytest.fixture(autouse=True)
def setup_db(memory_engine):
"""自动初始化数据库（每个测试函数前执行）"""
BaseModel.metadata.create_all(bind=memory_engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=memory_engine)
session_scope = scoped_session(SessionLocal)
CoreModel.query = session_scope.query_property()
yield session_scope
session_scope.remove()
@pytest.fixture
def app(setup_db):
"""创建完整的测试应用实例
注意：下面的 import 路径需要替换为你自己项目的实际路径，例如：
- from myproject.main import create_app
- from app.main import create_app
"""
from app.main import create_app
test_app = create_app()
return test_app
@pytest.fixture
def client(app) -> TestClient:
"""未认证的测试客户端"""
return TestClient(app)
@pytest.fixture
def auth_client(app) -> TestClient:
"""已认证的测试客户端（携带 JWT Token）"""
client = TestClient(app)
return client
@pytest.fixture
def sample_user(setup_db):
"""创建测试用户
注意：下面的 import 路径需要替换为你自己项目的用户模型路径
"""
from app.domain.auth.model.user import User
user = User(username="testuser", email="test@example.com")
user.set_password("Test123456")
user.add(True)
return user
5. 测试文件编写模板
5.1 ORM 模型测试（通用，适用于 yweb-core 和任何上层应用）
"""UserModel 模块测试
测试用户模型的 CRUD、软删除、查询等功能
"""
import pytest
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import StaticPool
from yweb.orm import CoreModel, BaseModel
class UserModel(BaseModel):
"""测试用户模型"""
__tablename__ = "test_users"
__table_args__ = {'extend_existing': True}
email = Column(String(200), comment="邮箱")
class TestUserModel:
"""用户模型 CRUD 测试"""
@pytest.fixture(autouse=True)
def setup_db(self, memory_engine):
"""自动初始化数据库会话
标准模式：创建表 → 创建会话 → 设置 query → yield → 清理
"""
BaseModel.metadata.create_all(bind=memory_engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=memory_engine)
session_scope = scoped_session(SessionLocal)
CoreModel.query = session_scope.query_property()
yield
session_scope.remove()
def test_create_user(self):
"""测试创建用户"""
user = UserModel(name="Alice", email="alice@example.com")
user.add(True)
assert user.id is not None
assert user.name == "Alice"
assert user.email == "alice@example.com"
def test_update_user(self):
"""测试更新用户"""
user = UserModel(name="Bob", email="bob@example.com")
user.add(True)
user.email = "bob_new@example.com"
user.update(True)
found = UserModel.get(user.id)
assert found.email == "bob_new@example.com"
def test_soft_delete_user(self):
"""测试软删除用户"""
user = UserModel(name="Charlie", email="charlie@example.com")
user.add(True)
user_id = user.id
user.delete(True)
assert UserModel.get(user_id) is None
def test_query_filter(self):
"""测试条件查询"""
UserModel(name="User1", email="u1@example.com").add(True)
UserModel(name="User2", email="u2@example.com").add(True)
found = UserModel.query.filter(UserModel.name == "User1").first()
assert found is not None
assert found.email == "u1@example.com"
5.2 API 层测试（上层应用 - import 路径需替换为你的项目）
"""角色管理 API 测试
测试 /api/v1/roles 相关端点
"""
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from unittest.mock import Mock, patch
from yweb.response import OK, BadRequest, NotFound
class TestRoleAPI:
"""角色 API 端点测试"""
@pytest.fixture
def app(self):
"""创建含角色路由的测试应用"""
from app.api.v1.role import create_role_router
test_app = FastAPI()
role_router = create_role_router()
test_app.include_router(role_router, prefix="/api/v1/roles")
return test_app
@pytest.fixture
def client(self, app):
return TestClient(app)
def test_list_roles(self, client):
"""测试获取角色列表"""
response = client.get("/api/v1/roles")
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
def test_create_role_success(self, client):
"""测试创建角色 - 正常情况"""
payload = {"name": "管理员", "code": "admin", "description": "系统管理员"}
response = client.post("/api/v1/roles/create", json=payload)
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert data["data"]["code"] == "admin"
def test_create_role_duplicate_code(self, client):
"""测试创建角色 - 编码重复应返回 409"""
payload = {"name": "管理员", "code": "admin"}
client.post("/api/v1/roles/create", json=payload)
response = client.post("/api/v1/roles/create", json=payload)
assert response.status_code == 409
def test_delete_system_role_forbidden(self, client):
"""测试删除系统角色 - 应被禁止"""
response = client.delete("/api/v1/roles/system_admin")
assert response.status_code in [400, 403]
5.3 Service 层测试（上层应用 - import 路径需替换为你的项目）
"""认证应用服务测试
测试 AuthApplicationService 的业务编排逻辑
"""
import pytest
from unittest.mock import Mock, patch, MagicMock
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import StaticPool
from yweb.orm import CoreModel, BaseModel
class TestAuthApplicationService:
"""认证服务测试"""
@pytest.fixture(autouse=True)
def setup_db(self, memory_engine):
"""初始化测试数据库"""
BaseModel.metadata.create_all(bind=memory_engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=memory_engine)
session_scope = scoped_session(SessionLocal)
CoreModel.query = session_scope.query_property()
yield
session_scope.remove()
@pytest.fixture
def auth_service(self):
"""创建认证服务实例"""
from app.services.auth_app import AuthApplicationService
return AuthApplicationService()
def test_register_user_success(self, auth_service):
"""测试用户注册 - 正常流程"""
result = auth_service.register(
username="newuser",
password="Secure123!",
email="new@example.com"
)
assert result is not None
assert result.username == "newuser"
def test_register_duplicate_username(self, auth_service):
"""测试用户注册 - 用户名重复"""
auth_service.register(username="existuser", password="Pass123!", email="a@b.com")
with pytest.raises(ValueError, match="已存在"):
auth_service.register(username="existuser", password="Pass456!", email="c@d.com")
def test_change_password_wrong_old_password(self, auth_service):
"""测试修改密码 - 旧密码错误"""
user = auth_service.register(username="pwduser", password="OldPass1!", email="p@q.com")
with pytest.raises(ValueError, match="密码"):
auth_service.change_password(
user_id=user.id,
old_password="WrongOld!",
new_password="NewPass1!"
)
5.4 Domain 层测试（领域服务 - import 路径需替换为你的项目）
"""角色领域服务测试
测试 RoleService 的业务规则
"""
import pytest
from unittest.mock import Mock
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import StaticPool
from yweb.orm import CoreModel, BaseModel
class TestRoleService:
"""角色服务测试"""
@pytest.fixture(autouse=True)
def setup_db(self, memory_engine):
"""初始化测试数据库"""
BaseModel.metadata.create_all(bind=memory_engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=memory_engine)
session_scope = scoped_session(SessionLocal)
CoreModel.query = session_scope.query_property()
yield
session_scope.remove()
@pytest.fixture
def role_service(self):
"""创建角色服务实例"""
from app.domain.auth.role_service import RoleService
return RoleService(role_model=RoleModel, user_model=UserModel)
def test_create_role(self, role_service):
"""测试创建角色"""
role = role_service.create_role(name="编辑", code="editor")
assert role.code == "editor"
def test_create_role_duplicate_code_raises(self, role_service):
"""测试创建重复编码角色应抛出 ValueError"""
role_service.create_role(name="管理员", code="admin")
with pytest.raises(ValueError, match="已存在"):
role_service.create_role(name="管理员2", code="admin")
def test_delete_system_role_raises(self, role_service):
"""测试删除系统角色应被禁止"""
role = role_service.create_role(name="超级管理员", code="super_admin", is_system=True)
with pytest.raises(ValueError, match="系统角色"):
role_service.delete_role(role.id)
6. 核心 Fixture 模式
6.1 数据库 Setup（最常用，必须掌握）
这是 yweb ORM 测试的标准模式，几乎每个测试文件都需要：
@pytest.fixture(autouse=True)
def setup_db(self, memory_engine):
"""标准数据库 setup fixture
步骤：
1. 创建所有表
2. 创建会话工厂
3. 创建 scoped_session
4. 设置 CoreModel.query（使 Active Record 方法工作）
5. yield（测试执行）
6. 清理会话
"""
BaseModel.metadata.create_all(bind=memory_engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=memory_engine)
session_scope = scoped_session(SessionLocal)
CoreModel.query = session_scope.query_property()
yield
session_scope.remove()
关键点：
memory_engine 来自根 conftest.py（使用 StaticPool）autouse=True 使其对类内所有测试方法自动生效- 必须设置
CoreModel.query，否则 .add() / .get() / .query 等 Active Record 方法无法工作
yield 后调用 session_scope.remove() 清理
6.2 测试模型定义
class UserModel(BaseModel):
"""测试用户模型"""
__tablename__ = "test_users"
__table_args__ = {'extend_existing': True}
email = Column(String(200), comment="邮箱")
关键点：
__tablename__ 使用 test_ 前缀，避免与生产表名冲突- 必须设置
__table_args__ = {'extend_existing': True}，防止 pytest 多文件加载时重复定义错误
- 类名不能以
Test 开头
6.3 MockUser / 工厂 Fixture
@pytest.fixture
def mock_user():
"""模拟用户对象工厂"""
class MockUser:
def __init__(self, id=1, username="testuser", email="test@example.com", is_active=True):
self.id = id
self.username = username
self.email = email
self.is_active = is_active
return MockUser
@pytest.fixture
def sample_user(setup_db):
"""创建实际的测试用户（写入数据库）"""
user = UserModel(name="测试用户", email="test@example.com")
user.add(True)
return user
7. Mock 策略
7.1 推荐方式优先级
unittest.mock（Mock / MagicMock / AsyncMock / patch） — 项目主要使用方式pytest-mock（mocker fixture） — 简单场景可选用- 自定义 Mock 类（通过 fixture） — 复杂对象推荐
7.2 使用 unittest.mock
from unittest.mock import Mock, patch, MagicMock, AsyncMock
class TestSyncService:
"""同步服务测试"""
def test_sync_calls_api(self):
"""测试同步操作调用外部 API"""
with patch("app.domain.xxx.client.XxxClient.get_departments") as mock_api:
mock_api.return_value = [{"id": 1, "name": "技术部"}]
service = SyncService()
result = service.sync_departments()
mock_api.assert_called_once()
assert len(result) == 1
def test_service_with_mock_dependency(self):
"""使用 Mock 替代依赖"""
mock_repo = Mock()
mock_repo.find_by_id.return_value = SampleRole(id=1, name="admin")
service = RoleService(role_model=mock_repo)
role = service.get_role(1)
assert role.name == "admin"
mock_repo.find_by_id.assert_called_once_with(1)
7.3 使用 mocker fixture（pytest-mock）
def test_cache_hit(self, mocker):
"""测试缓存命中"""
mock_get = mocker.patch("yweb.cache.backend.get")
mock_get.return_value = '{"id": 1, "name": "cached"}'
result = get_user_cached(1)
assert result["name"] == "cached"
mock_get.assert_called_once()
7.4 异步 Mock
from unittest.mock import AsyncMock
@pytest.mark.asyncio
async def test_async_job_execution(self):
"""测试异步任务执行"""
mock_handler = AsyncMock(return_value="done")
result = await execute_job(mock_handler)
assert result == "done"
mock_handler.assert_awaited_once()
8. 异步测试
8.1 基本异步测试
import pytest
@pytest.mark.asyncio
async def test_async_operation():
"""测试异步操作"""
result = await some_async_function()
assert result is not None
8.2 异步测试类
class TestAsyncService:
"""异步服务测试"""
@pytest.mark.asyncio
async def test_fetch_data(self):
"""测试异步数据获取"""
service = AsyncDataService()
data = await service.fetch()
assert len(data) > 0
8.3 pytest-asyncio 配置
如果整个文件都是异步测试，可在文件头部声明：
import pytest
pytestmark = pytest.mark.asyncio
9. 断言规范
9.1 基本断言
assert user.id is not None
assert user.name == "Alice"
assert isinstance(user.id, int)
assert len(users) == 3
assert "admin" in role_codes
assert user.ver == 1, f"初始版本号应为 1，实际为 {user.ver}"
assert user.updated_at is None, "首次添加时 updated_at 应为空"
9.2 异常断言
with pytest.raises(ValueError):
service.create_role(name="", code="")
with pytest.raises(ValueError, match="已存在"):
service.create_role(name="dup", code="admin")
with pytest.raises(ValueError) as exc_info:
service.delete_system_role(role_id)
assert "系统角色" in str(exc_info.value)
9.3 响应断言（API 测试）
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert data["message"] == "操作成功"
assert data["data"]["id"] is not None
assert data["status"] == "error"
assert "不存在" in data["message"]
10. 参数化测试
10.1 @pytest.mark.parametrize
@pytest.mark.parametrize("email,is_valid", [
("user@example.com", True),
("user@test.org", True),
("invalid-email", False),
("", False),
(None, False),
])
def test_email_validation(self, email, is_valid):
"""测试邮箱校验"""
if is_valid:
user = UserModel(name="Test", email=email)
user.add(True)
assert user.email == email
else:
with pytest.raises((ValueError, TypeError)):
validate_email(email)
@pytest.mark.parametrize("role_code,should_protect", [
("super_admin", True),
("admin", True),
("custom_role", False),
])
def test_system_role_protection(self, role_code, should_protect):
"""测试系统角色保护机制"""
role = create_role(code=role_code, is_system=should_protect)
if should_protect:
with pytest.raises(ValueError, match="系统角色"):
role_service.delete_role(role.id)
else:
role_service.delete_role(role.id)
11. 测试标记
11.1 常用标记
@pytest.mark.unit
@pytest.mark.integration
@pytest.mark.slow
@pytest.mark.asyncio
11.2 按标记运行
pytest -m unit
pytest -m "not slow"
pytest -m "integration"
12. 测试辅助工具（helpers/）
12.1 设计原则
- helpers 目录放非测试文件，提供辅助函数和工具
- 避免在核心代码中添加测试专用方法
- 通过
__init__.py 导出公共接口
12.2 示例结构
"""测试数据工厂"""
def create_test_user(name="测试用户", email="test@example.com", **kwargs):
"""创建测试用户"""
from app.domain.auth.model.user import User
user = User(name=name, email=email, **kwargs)
user.set_password("DefaultPass123!")
user.add(True)
return user
def create_test_role(name="测试角色", code="test_role", **kwargs):
"""创建测试角色"""
from app.domain.permission.entities import Role
role = Role(name=name, code=code, **kwargs)
role.add(True)
return role
"""认证测试辅助"""
def create_test_token(user_id: int, username: str = "testuser") -> str:
"""生成测试用 JWT Token"""
from yweb.auth import JWTManager
manager = JWTManager(secret_key="test-secret", algorithm="HS256")
payload = {"sub": username, "user_id": user_id}
return manager.create_access_token(payload)
def get_auth_headers(user_id: int = 1) -> dict:
"""获取带认证的请求头"""
token = create_test_token(user_id)
return {"Authorization": f"Bearer {token}"}
13. 覆盖率
13.1 运行覆盖率
pytest --cov=yweb --cov-report=term-missing
pytest --cov=yweb --cov-report=html
pytest --cov=yweb --cov-fail-under=80
13.2 覆盖率关注重点
| 优先级 | 模块 | 目标 |
|---|
| 高 | Domain 层（领域模型 + 领域服务） | ≥ 90% |
| 高 | Service 层（业务编排） | ≥ 85% |
| 中 | API 层（端点） | ≥ 80% |
| 中 | ORM 模型（自定义方法） | ≥ 80% |
| 低 | 配置/启动代码 | ≥ 60% |
14. 常见反模式（避免）
14.1 禁止的做法
class TestUser(unittest.TestCase):
def setUp(self):
...
def test_user_crud_and_role_assignment():
user = create_user(...)
assert user.id
assign_role(user, "admin")
assert user.has_role("admin")
class TestUserFactory:
...
def test_a_create():
global user_id
user_id = create_user().id
def test_b_get():
get_user(user_id)
engine = create_engine("sqlite:///test.db")
from yweb.orm import init_database
init_database(...)
14.2 推荐的做法
class TestUser:
@pytest.fixture(autouse=True)
def setup_db(self, memory_engine):
...
def test_create_user(self):
...
def test_assign_role_to_user(self):
...
class UserFactory:
...
def test_get_user(self):
user = UserModel(name="Test").add(True)
found = UserModel.get(user.id)
assert found is not None
@pytest.fixture(scope="function")
def memory_engine():
...
15. DDD 分层测试策略总览
┌──────────────────────────────────────────────────┐
│ API 层测试 (test_api/) │
│ - 使用 TestClient 发 HTTP 请求 │
│ - 验证状态码、响应格式、DTO 转换 │
│ - Mock Service 层（或用真实 Service + 测试 DB） │
├──────────────────────────────────────────────────┤
│ Service 层测试 (test_services/) │
│ - 测试业务编排逻辑 │
│ - 可 Mock Domain 层依赖，也可用真实 DB │
│ - 验证事务行为、跨聚合协调 │
├──────────────────────────────────────────────────┤
│ Domain 层测试 (test_domain/) │
│ - 测试领域服务的业务规则 │
│ - 测试领域模型的行为方法 │
│ - 使用内存 DB + Active Record │
│ - 验证校验逻辑、状态变更、异常抛出 │
├──────────────────────────────────────────────────┤
│ ORM 层测试 (test_orm/) │
│ - 测试自定义 ORM 功能 │
│ - CRUD、软删除、分页、关系等 │
│ - 使用内存 SQLite + setup_db fixture │
└──────────────────────────────────────────────────┘
16. 快速检查清单
编写测试文件前，对照此清单：
