菜单
YWeb 框架 Pytest 单元测试编写规范。适用于 yweb-core 底层框架本身，以及通过 pip install yweb 安装后基于该框架开发的任何上层应用。涵盖测试目录结构、conftest 编写、fixture 模式、ORM 测试、API 测试、Service 测试、Domain 测试、Mock 策略、异步测试等场景。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
基于 SOC 职业分类
| name | yweb-testing |
| description | YWeb 框架 Pytest 单元测试编写规范。适用于 yweb-core 底层框架本身，以及通过 pip install yweb 安装后基于该框架开发的任何上层应用。涵盖测试目录结构、conftest 编写、fixture 模式、ORM 测试、API 测试、Service 测试、Domain 测试、Mock 策略、异步测试等场景。 |
本 Skill 是 AI 编程助手为 yweb 框架生态编写 pytest 测试文件的权威指南。适用于两种场景：
pip install yweb 安装框架后，测试应用自身的 API 层、Service 层、Domain 层业务逻辑关于上层应用：本文档中的"上层应用"指任何通过
pip install yweb引入框架后独立开发的项目。文档中以虚构的示例项目my-app演示目录结构和代码模板，实际项目名称和结构请根据自身项目调整。上层应用看不到 yweb-core 的源码和测试文件，只能通过from yweb import ...使用框架提供的公共 API。
核心原则：测试代码的质量标准与生产代码一致。测试应快速、可靠、可维护、易读。
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/
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
test_ 前缀命名，对应被测模块unit/ 和 integration/ 子目录helpers/ 目录放测试辅助工具（不是测试文件）__init__.py[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
# requirements-dev.txt
pytest>=8.0
pytest-asyncio>=0.23
pytest-mock>=3.12
pytest-cov>=4.1
httpx>=0.27 # 异步 TestClient 支持
test_<模块名>.py # 例：test_base_model.py、test_role_api.py
class TestBaseModel: # 测试类：Test + 被测类/模块名
class TestRoleService: # 测试类：Test + 被测服务名
class TestRoleAPI: # 测试类：Test + 被测 API 名
def test_create_user_with_valid_email(self): # test_ + 动作 + 条件/场景
def test_delete_role_raises_error_for_system_role(self):
def test_get_user_returns_none_when_not_found(self):
严重警告：pytest 的测试发现机制会自动将
test_*.py文件中所有以Test开头的 class 识别为测试类，并尝试收集其中以test_开头的方法作为测试用例。如果辅助类、实体类、模型类以Test开头，pytest 会产生警告或错误。
规则：只有真正的测试类才以 Test 开头，其他所有类一律禁止。
这条规则适用于测试文件中定义的所有非测试类，包括但不限于：
# ✅ 正确 — 非测试类不以 Test 开头
class UserModel(BaseModel): # 测试用 ORM 模型
class MockUser: # Mock 类
class SampleRole: # 示例实体
class RoleFactory: # 工厂类
class FakeEmailService: # Fake 类
# ❌ 错误 — pytest 会误识别为测试类并发出警告
class TestUserModel(BaseModel): # ← 会被 pytest 当作测试类！
class TestHelper: # ← 会被 pytest 当作测试类！
class TestData: # ← 会被 pytest 当作测试类！
记忆口诀：文件名
test_开头，测试类Test开头，测试方法test_开头——只有这三者可以用 test 前缀，其余一切类和函数都不能以 test/Test 开头。
class TestBaseModel:
"""BaseModel 基础功能测试"""
def test_model_has_id(self):
"""测试模型创建后自动分配 ID"""
...
def test_soft_delete_restore(self):
"""测试软删除后恢复记录"""
...
"""
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
# ==================== Pytest Hook：全局初始化 ====================
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()
# ==================== 数据库 Fixtures ====================
@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()
# ==================== FastAPI Fixtures ====================
@pytest.fixture
def app():
"""创建测试用 FastAPI 应用"""
test_app = FastAPI(title="Test App")
return test_app
@pytest.fixture
def client(app):
"""创建测试客户端"""
return TestClient(app)
# ==================== 临时文件 Fixtures ====================
@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)
"""
应用层 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
# yweb 框架公共 API（通过 pip install yweb 安装）
from yweb.orm import CoreModel, BaseModel
# ==================== 数据库 Fixtures ====================
@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()
# ==================== 应用 Fixtures ====================
@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)
# 创建测试用户并获取 Token
# token = create_test_token(user_id=1, username="testadmin")
# client.headers["Authorization"] = f"Bearer {token}"
return client
# ==================== 测试数据 Fixtures ====================
@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
"""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"
"""角色管理 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]
"""认证应用服务测试
测试 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!"
)
"""角色领域服务测试
测试 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 # ← 替换为你的领域服务模块
# 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)
这是 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() 清理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 开头@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
unittest.mock（Mock / MagicMock / AsyncMock / patch） — 项目主要使用方式pytest-mock（mocker fixture） — 简单场景可选用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)
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()
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()
import pytest
@pytest.mark.asyncio
async def test_async_operation():
"""测试异步操作"""
result = await some_async_function()
assert result is not None
class TestAsyncService:
"""异步服务测试"""
@pytest.mark.asyncio
async def test_fetch_data(self):
"""测试异步数据获取"""
service = AsyncDataService()
data = await service.fetch()
assert len(data) > 0
如果整个文件都是异步测试，可在文件头部声明：
import pytest
pytestmark = pytest.mark.asyncio # 整个文件的测试方法都标记为 asyncio
# 值断言
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 应为空"
# 断言抛出异常
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)
# HTTP 状态码
assert response.status_code == 200
# 响应体结构（yweb 统一响应格式）
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"]
@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) # 应成功
@pytest.mark.unit # 单元测试
@pytest.mark.integration # 集成测试
@pytest.mark.slow # 慢速测试
@pytest.mark.asyncio # 异步测试
pytest -m unit # 只跑单元测试
pytest -m "not slow" # 跳过慢速测试
pytest -m "integration" # 只跑集成测试
__init__.py 导出公共接口# tests/helpers/factory.py
"""测试数据工厂"""
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
# tests/helpers/auth_utils.py
"""认证测试辅助"""
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}"}
# 基础覆盖率
pytest --cov=yweb --cov-report=term-missing
# HTML 报告
pytest --cov=yweb --cov-report=html
# 指定最低覆盖率
pytest --cov=yweb --cov-fail-under=80
| 优先级 | 模块 | 目标 |
|---|---|---|
| 高 | Domain 层（领域模型 + 领域服务） | ≥ 90% |
| 高 | Service 层（业务编排） | ≥ 85% |
| 中 | API 层（端点） | ≥ 80% |
| 中 | ORM 模型（自定义方法） | ≥ 80% |
| 低 | 配置/启动代码 | ≥ 60% |
# ❌ 使用 unittest.TestCase（应使用纯 pytest 风格）
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")
# ❌ 辅助类以 Test 开头
class TestUserFactory: # pytest 会误识别
...
# ❌ 测试依赖执行顺序
def test_a_create(): # 不要依赖 test_a 在 test_b 之前执行
global user_id
user_id = create_user().id
def test_b_get():
get_user(user_id) # 依赖全局变量
# ❌ 硬编码数据库连接（应使用 fixture）
engine = create_engine("sqlite:///test.db") # 不要在模块顶层这样做
# ❌ 在测试中直接导入并调用 init_database
from yweb.orm import init_database
init_database(...) # 使用 fixture 管理，不要直接调用
# ✅ 纯 pytest 风格
class TestUser:
@pytest.fixture(autouse=True)
def setup_db(self, memory_engine):
...
# ✅ 单一职责
def test_create_user(self):
...
def test_assign_role_to_user(self):
...
# ✅ 辅助类不以 Test 开头
class UserFactory:
...
# ✅ 测试之间无依赖，每个测试自给自足
def test_get_user(self):
user = UserModel(name="Test").add(True) # 自己创建数据
found = UserModel.get(user.id)
assert found is not None
# ✅ 使用 fixture 管理数据库
@pytest.fixture(scope="function")
def memory_engine():
...
┌──────────────────────────────────────────────────┐
│ 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 │
└──────────────────────────────────────────────────┘
编写测试文件前，对照此清单：
test_ 开头Test 开头，辅助类不以 Test 开头test_ 开头，名称描述测试场景memory_engine fixture（来自 conftest.py）setup_db autouse fixtureextend_existing=Truetest_ 前缀@pytest.mark.asynciopytest.raisesYWeb 基础设施模块规范。在使用缓存（@cached）、异常处理（Err/register_exception_handlers）、日志（get_logger）、配置（AppSettings/YAML）、文件存储（本地/OSS/S3）、定时任务（Scheduler）、限流（setup_ratelimit）时使用。
YWeb DDD 分层架构与 API 设计规范。在创建或修改 API 路由、Service 层、领域模型、DTO 时使用。涵盖瘦 API 原则、服务层拆分、Model 设计、DTO 转换、响应格式等。
YWeb ORM 使用规范。在编写数据库模型定义、CRUD 操作、查询过滤、分页、软删除、事务管理、批量操作、关系定义等数据层代码时使用。基于 SQLAlchemy 的 Active Record 模式。
YWeb 认证授权与权限管理规范。在实现用户登录、JWT Token 处理、权限校验、角色管理、组织架构、OAuth 集成等功能时使用。
YWeb 框架编码总规范。基于 FastAPI + SQLAlchemy 的 Web 框架，采用 Active Record + DDD 分层架构。在编写任何基于 yweb 框架的代码时使用此技能，包括 API 开发、ORM 操作、认证授权、缓存、异常处理等场景。
审查和改进 pytest 测试的真实性，防止 Happy Path、浅层、实现驱动、同义反复等虚假测试；在测试失败时执行先诊断后改码流程。用于测试评审、测试重构、失败排查场景。