From 61f2cec1d1abe936c53eb9f77b8bc748e619ebb6 Mon Sep 17 00:00:00 2001 From: helloway Date: Tue, 10 Feb 2026 12:40:35 +0800 Subject: [PATCH] feat: configure pytest framework MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 配置pytest-asyncio模式 2. 同步pyproject.toml中的pytest配置 3. 验证conftest.py中的所有fixtures 4. 生成pytest配置总结文档 5. 创建backend/tests/test_example.py 6. 创建backend/verify_test_config.py --- backend/pyproject.toml | 37 +++ backend/pytest.ini | 4 + backend/tests/test_example.py | 593 ++++++++++++++++++++++++++++++++++ backend/verify_test_config.py | 453 ++++++++++++++++++++++++++ 4 files changed, 1087 insertions(+) create mode 100644 backend/tests/test_example.py create mode 100644 backend/verify_test_config.py diff --git a/backend/pyproject.toml b/backend/pyproject.toml index eab1a95..5e24fc8 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -70,11 +70,48 @@ testpaths = ["tests"] python_files = ["test_*.py"] python_classes = ["Test*"] python_functions = ["test_*"] +# Asyncio configuration +asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" addopts = [ "-v", + "-l", + "-ra", "--strict-markers", "--strict-config", "--cov=app", "--cov-report=term-missing", "--cov-report=html", + "--cov-branch", + "-p no:warnings", ] +markers = [ + "slow: marks tests as slow (deselect with '-m \"not slow\"')", + "integration: marks tests as integration tests", + "unit: marks tests as unit tests", + "api: marks tests as API endpoint tests", + "auth: marks tests related to authentication", + "community: marks tests related to communities", + "content: marks tests related to content management", + "publish: marks tests related to publishing", + "analytics: marks tests related to analytics", +] + +[tool.coverage.run] +source = ["app"] +omit = [ + "*/tests/*", + "*/migrations/*", + "*/__pycache__/*", + "*/venv/*", + "*/.venv/*", +] + +[tool.coverage.report] +fail_under = 80 +precision = 2 +show_missing = true +skip_covered = false + +[tool.coverage.html] +directory = "htmlcov" diff --git a/backend/pytest.ini b/backend/pytest.ini index c67a46a..4b3dd31 100644 --- a/backend/pytest.ini +++ b/backend/pytest.ini @@ -9,6 +9,10 @@ python_functions = test_* # Test directories testpaths = tests +# Asyncio configuration +asyncio_mode = auto +asyncio_default_fixture_loop_scope = function + # Output options addopts = # Verbose output diff --git a/backend/tests/test_example.py b/backend/tests/test_example.py new file mode 100644 index 0000000..a0524ac --- /dev/null +++ b/backend/tests/test_example.py @@ -0,0 +1,593 @@ +""" +测试示例 - 演示如何使用各种 fixtures + +这个文件包含了使用 conftest.py 中定义的各种 fixtures 的示例。 +可以作为编写新测试时的参考模板。 +""" + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models import User, Community +from app.models.content import Content +from app.core.security import verify_password, create_access_token + + +# ============================================================================== +# 数据库 Fixtures 示例 +# ============================================================================== + + +def test_db_session_example(db_session: Session): + """ + 示例:使用 db_session fixture + + db_session 提供一个数据库会话,测试结束后自动回滚所有更改。 + 每个测试函数都有独立的会话,确保测试隔离。 + """ + # 在测试数据库中创建新用户 + new_user = User( + username="example_user", + email="example@test.com", + hashed_password="hashed_password_here", + full_name="Example User", + is_active=True, + ) + db_session.add(new_user) + db_session.commit() + db_session.refresh(new_user) + + # 验证用户已创建 + assert new_user.id is not None + assert new_user.username == "example_user" + + # 测试结束后,这个用户会被自动删除(事务回滚) + + +def test_db_session_query_example(db_session: Session): + """ + 示例:在 db_session 中查询数据 + + 由于测试隔离,这个测试看不到上一个测试创建的用户。 + """ + # 查询所有用户 + users = db_session.query(User).all() + + # 由于测试隔离,这里应该是空的(或只有 fixture 创建的用户) + # 上一个测试创建的用户已经回滚 + for user in users: + print(f"Found user: {user.username}") + + +# ============================================================================== +# FastAPI TestClient Fixture 示例 +# ============================================================================== + + +def test_client_basic_example(client: TestClient): + """ + 示例:使用 client fixture 测试 API 端点 + + client 是 FastAPI 的 TestClient,自动注入测试数据库会话。 + """ + # 测试健康检查端点(不需要认证) + response = client.get("/api/health") + + # 验证响应 + assert response.status_code == 200 + data = response.json() + assert data["status"] == "healthy" + + +def test_client_post_example(client: TestClient): + """ + 示例:使用 client 测试 POST 请求 + """ + # 发送 POST 请求 + response = client.post( + "/api/some-endpoint", + json={"key": "value"}, + ) + + # 根据实际 API 验证响应 + # assert response.status_code == 200 + + +# ============================================================================== +# 测试数据 Fixtures 示例 +# ============================================================================== + + +def test_test_user_example(test_user: User): + """ + 示例:使用 test_user fixture + + test_user 是预先创建的测试用户,无需手动创建。 + """ + # 验证测试用户属性 + assert test_user.username == "testuser" + assert test_user.email == "testuser@example.com" + assert test_user.is_active is True + assert test_user.is_superuser is False + + # 验证密码(原始密码是 "testpass123") + assert verify_password("testpass123", test_user.hashed_password) + + +def test_test_superuser_example(test_superuser: User): + """ + 示例:使用 test_superuser fixture + + test_superuser 是预先创建的超级管理员用户。 + """ + assert test_superuser.username == "admin" + assert test_superuser.is_superuser is True + + # 超级管理员有特殊权限 + # 可以用于测试需要管理员权限的功能 + + +def test_test_community_example(test_community: Community): + """ + 示例:使用 test_community fixture + + test_community 是预先创建的测试社区。 + """ + assert test_community.name == "Test Community" + assert test_community.slug == "test-community" + assert test_community.is_active is True + + +def test_combined_fixtures_example( + db_session: Session, + test_user: User, + test_community: Community, +): + """ + 示例:组合使用多个 fixtures + + 可以在同一个测试中使用多个 fixtures。 + """ + # 使用测试用户和社区创建内容 + content = Content( + title="Example Content", + body="This is example content", + author_id=test_user.id, + community_id=test_community.id, + ) + db_session.add(content) + db_session.commit() + db_session.refresh(content) + + # 验证内容 + assert content.id is not None + assert content.author_id == test_user.id + assert content.community_id == test_community.id + + +# ============================================================================== +# 认证 Fixtures 示例 +# ============================================================================== + + +def test_user_token_example(user_token: str): + """ + 示例:使用 user_token fixture + + user_token 是 test_user 的 JWT token。 + """ + # token 是一个字符串 + assert isinstance(user_token, str) + assert len(user_token) > 0 + + # 可以手动构造 headers + headers = {"Authorization": f"Bearer {user_token}"} + # 然后用于 API 请求 + + +def test_auth_headers_example(client: TestClient, auth_headers: dict): + """ + 示例:使用 auth_headers fixture 测试需要认证的端点 + + auth_headers 包含 Authorization header 和 X-Community-Id header。 + """ + # 测试需要认证的端点 + response = client.get("/api/auth/me", headers=auth_headers) + + # 应该返回当前用户信息 + assert response.status_code == 200 + data = response.json() + assert data["user"]["username"] == "testuser" + + +def test_superuser_auth_headers_example( + client: TestClient, + superuser_auth_headers: dict, +): + """ + 示例:使用 superuser_auth_headers 测试管理员端点 + """ + # 测试只有超级管理员才能访问的端点 + response = client.get("/api/auth/users", headers=superuser_auth_headers) + + # 应该成功返回用户列表 + assert response.status_code == 200 + users = response.json() + assert isinstance(users, list) + + +# ============================================================================== +# 测试隔离示例 +# ============================================================================== + + +def test_isolation_test_1(db_session: Session, test_user: User): + """ + 示例:测试隔离 - 第一个测试 + + 这个测试修改了 test_user,但不会影响其他测试。 + """ + # 修改用户全名 + original_name = test_user.full_name + test_user.full_name = "Modified Name" + db_session.commit() + + # 在这个测试中,更改是可见的 + db_session.refresh(test_user) + assert test_user.full_name == "Modified Name" + + # 测试结束后会回滚 + + +def test_isolation_test_2(test_user: User): + """ + 示例:测试隔离 - 第二个测试 + + 即使上一个测试修改了 test_user,这个测试看到的是原始数据。 + """ + # 由于事务回滚,这里看到的是原始值 + assert test_user.full_name == "Test User" + # 不是上一个测试中的 "Modified Name" + + +# ============================================================================== +# 参数化测试示例 +# ============================================================================== + + +@pytest.mark.parametrize( + "username,email,is_valid", + [ + ("user1", "user1@example.com", True), + ("user2", "user2@example.com", True), + ("user3", "user3@example.com", True), + ("", "invalid@example.com", False), # 空用户名 + ("user4", "", False), # 空邮箱 + ], +) +def test_parametrized_user_creation_example( + db_session: Session, + username: str, + email: str, + is_valid: bool, +): + """ + 示例:参数化测试 + + 使用 @pytest.mark.parametrize 可以用不同参数运行同一个测试。 + 这个测试会运行 5 次,每次使用不同的参数。 + """ + if is_valid: + # 有效输入应该成功创建用户 + user = User( + username=username, + email=email, + hashed_password="test_hash", + full_name="Test User", + ) + db_session.add(user) + db_session.commit() + assert user.id is not None + else: + # 无效输入应该抛出异常 + with pytest.raises(Exception): + user = User( + username=username, + email=email, + hashed_password="test_hash", + full_name="Test User", + ) + db_session.add(user) + db_session.commit() + + +# ============================================================================== +# 测试标记示例 +# ============================================================================== + + +@pytest.mark.unit +def test_marked_as_unit(): + """ + 示例:标记为单元测试 + + 运行方式:pytest -m unit + """ + assert 1 + 1 == 2 + + +@pytest.mark.api +@pytest.mark.auth +def test_marked_as_api_and_auth(client: TestClient): + """ + 示例:多个标记 + + 运行方式:pytest -m "api and auth" + """ + # 测试 API 认证端点 + response = client.get("/api/auth/status") + assert response.status_code == 200 + + +@pytest.mark.slow +def test_marked_as_slow(): + """ + 示例:标记为慢速测试 + + 跳过运行:pytest -m "not slow" + """ + import time + time.sleep(0.1) # 模拟慢速操作 + assert True + + +@pytest.mark.integration +def test_marked_as_integration( + client: TestClient, + auth_headers: dict, + db_session: Session, + test_user: User, + test_community: Community, +): + """ + 示例:标记为集成测试 + + 集成测试通常涉及多个组件的交互。 + 运行方式:pytest -m integration + """ + # 1. 创建内容 + content = Content( + title="Integration Test Content", + body="Testing integration", + author_id=test_user.id, + community_id=test_community.id, + ) + db_session.add(content) + db_session.commit() + db_session.refresh(content) + + # 2. 通过 API 获取内容 + response = client.get( + f"/api/communities/{test_community.id}/content/{content.id}", + headers=auth_headers, + ) + + # 3. 验证完整流程 + assert response.status_code == 200 + data = response.json() + assert data["title"] == "Integration Test Content" + + +# ============================================================================== +# 异步测试示例 +# ============================================================================== + + +@pytest.mark.asyncio +async def test_async_example(): + """ + 示例:异步测试 + + 由于配置了 asyncio_mode = auto,可以直接编写异步测试。 + """ + # 模拟异步操作 + async def async_operation(): + return "result" + + result = await async_operation() + assert result == "result" + + +# ============================================================================== +# 异常测试示例 +# ============================================================================== + + +def test_exception_example(): + """ + 示例:测试异常 + + 使用 pytest.raises 验证代码应该抛出特定异常。 + """ + with pytest.raises(ValueError): + raise ValueError("Expected error") + + +def test_exception_with_match_example(): + """ + 示例:测试异常消息 + + 可以使用 match 参数验证异常消息。 + """ + with pytest.raises(ValueError, match="specific message"): + raise ValueError("specific message in the error") + + +# ============================================================================== +# Fixture 依赖示例 +# ============================================================================== + + +@pytest.fixture +def custom_content(db_session: Session, test_user: User, test_community: Community): + """ + 示例:自定义 fixture + + 可以创建依赖其他 fixtures 的自定义 fixture。 + """ + content = Content( + title="Custom Fixture Content", + body="Created by custom fixture", + author_id=test_user.id, + community_id=test_community.id, + ) + db_session.add(content) + db_session.commit() + db_session.refresh(content) + return content + + +def test_custom_fixture_example(custom_content: Content): + """ + 示例:使用自定义 fixture + """ + assert custom_content.title == "Custom Fixture Content" + assert custom_content.id is not None + + +# ============================================================================== +# 多社区隔离测试示例 +# ============================================================================== + + +def test_community_isolation_example( + client: TestClient, + auth_headers: dict, + another_user_auth_headers: dict, + test_community: Community, + test_another_community: Community, + db_session: Session, + test_user: User, + test_another_user: User, +): + """ + 示例:测试多社区之间的数据隔离 + + 确保一个社区的用户无法访问另一个社区的资源。 + """ + # test_user 在 test_community 中创建内容 + content = Content( + title="Community 1 Content", + body="This belongs to community 1", + author_id=test_user.id, + community_id=test_community.id, + ) + db_session.add(content) + db_session.commit() + db_session.refresh(content) + + # test_user 应该能访问自己社区的内容 + response = client.get( + f"/api/communities/{test_community.id}/content/{content.id}", + headers=auth_headers, + ) + assert response.status_code == 200 + + # test_another_user 不应该能访问其他社区的内容 + response = client.get( + f"/api/communities/{test_community.id}/content/{content.id}", + headers=another_user_auth_headers, + ) + # 应该返回 403 或 404(取决于实现) + assert response.status_code in [403, 404] + + +# ============================================================================== +# 实用技巧 +# ============================================================================== + + +def test_debugging_example(test_user: User): + """ + 示例:调试技巧 + + 运行测试时添加 -s 参数可以看到 print 输出: + pytest tests/test_example.py::test_debugging_example -s + """ + print(f"\nDebugging: test_user.id = {test_user.id}") + print(f"Debugging: test_user.username = {test_user.username}") + + # 在测试中添加 breakpoint() 可以进入调试器 + # breakpoint() + + assert test_user.username == "testuser" + + +def test_skip_example(): + """ + 示例:跳过测试 + + 使用 @pytest.mark.skip 可以跳过测试。 + """ + pytest.skip("Skipping this test for demonstration") + + +@pytest.mark.skipif(True, reason="Conditional skip") +def test_conditional_skip_example(): + """ + 示例:条件跳过 + + 根据条件决定是否跳过测试。 + """ + assert True + + +# ============================================================================== +# 最佳实践总结 +# ============================================================================== + +""" +测试编写最佳实践: + +1. 测试名称应该清晰描述测试内容 + ✅ test_user_login_with_valid_credentials + ❌ test_1 + +2. 每个测试应该测试一个具体功能 + ✅ 一个测试验证登录成功 + ✅ 另一个测试验证登录失败 + ❌ 一个测试验证登录的所有情况 + +3. 使用 fixtures 避免重复代码 + ✅ 使用 test_user fixture + ❌ 在每个测试中手动创建用户 + +4. 确保测试隔离 + ✅ 使用 db_session 的事务回滚 + ❌ 在测试之间共享数据库状态 + +5. 使用适当的断言 + ✅ assert response.status_code == 200 + ❌ assert response.status_code # 不清晰 + +6. 添加清晰的文档字符串 + ✅ 说明测试的目的和预期行为 + ❌ 没有文档或文档不清晰 + +7. 使用标记分类测试 + ✅ @pytest.mark.api, @pytest.mark.slow + ❌ 所有测试混在一起 + +8. 参数化相似的测试 + ✅ 使用 @pytest.mark.parametrize + ❌ 复制粘贴相似的测试函数 + +9. 测试边界条件和异常情况 + ✅ 测试空输入、超长输入、无效输入 + ❌ 只测试正常情况 + +10. 保持测试简洁 + ✅ 专注于测试逻辑 + ❌ 过多的设置代码 +""" diff --git a/backend/verify_test_config.py b/backend/verify_test_config.py new file mode 100644 index 0000000..b3365fd --- /dev/null +++ b/backend/verify_test_config.py @@ -0,0 +1,453 @@ +#!/usr/bin/env python3 +""" +Pytest 配置验证脚本 + +运行这个脚本来验证 pytest 测试框架的所有配置是否正确设置。 + +使用方法: + python verify_test_config.py + +返回值: + 0 - 所有检查通过 + 1 - 有检查失败 +""" + +import os +import sys +from pathlib import Path +from typing import List, Tuple +import importlib.util + + +# 颜色输出支持 +class Colors: + """终端颜色代码""" + GREEN = '\033[92m' + RED = '\033[91m' + YELLOW = '\033[93m' + BLUE = '\033[94m' + BOLD = '\033[1m' + END = '\033[0m' + + +def print_success(message: str): + """打印成功消息""" + print(f"{Colors.GREEN}✓{Colors.END} {message}") + + +def print_error(message: str): + """打印错误消息""" + print(f"{Colors.RED}✗{Colors.END} {message}") + + +def print_warning(message: str): + """打印警告消息""" + print(f"{Colors.YELLOW}⚠{Colors.END} {message}") + + +def print_info(message: str): + """打印信息消息""" + print(f"{Colors.BLUE}ℹ{Colors.END} {message}") + + +def print_header(message: str): + """打印标题""" + print(f"\n{Colors.BOLD}{message}{Colors.END}") + + +class TestConfigValidator: + """测试配置验证器""" + + def __init__(self): + self.project_root = Path(__file__).parent + self.passed_checks = 0 + self.failed_checks = 0 + self.warnings = 0 + + def check_file_exists(self, file_path: Path, description: str) -> bool: + """检查文件是否存在""" + if file_path.exists(): + print_success(f"{description}: {file_path.name}") + self.passed_checks += 1 + return True + else: + print_error(f"{description}: {file_path.name} 不存在") + self.failed_checks += 1 + return False + + def check_pytest_ini(self) -> bool: + """检查 pytest.ini 配置""" + print_header("1. 检查 pytest.ini 配置") + + pytest_ini = self.project_root / "pytest.ini" + if not self.check_file_exists(pytest_ini, "pytest.ini 文件"): + return False + + content = pytest_ini.read_text() + + # 检查必需的配置项 + checks = [ + ("testpaths", "测试目录配置"), + ("python_files", "测试文件模式"), + ("python_classes", "测试类模式"), + ("python_functions", "测试函数模式"), + ("asyncio_mode", "异步测试模式"), + ("asyncio_default_fixture_loop_scope", "异步 fixture 作用域"), + ("--cov=app", "覆盖率配置"), + ("--cov-report=", "覆盖率报告"), + ("--cov-branch", "分支覆盖率"), + ("fail_under = 80", "最低覆盖率要求 (80%)"), + ] + + for config_key, description in checks: + if config_key in content: + print_success(f"{description}: 已配置") + self.passed_checks += 1 + else: + print_error(f"{description}: 未配置") + self.failed_checks += 1 + + return True + + def check_pyproject_toml(self) -> bool: + """检查 pyproject.toml 配置""" + print_header("2. 检查 pyproject.toml 配置") + + pyproject = self.project_root / "pyproject.toml" + if not self.check_file_exists(pyproject, "pyproject.toml 文件"): + return False + + content = pyproject.read_text() + + # 检查必需的配置项 + checks = [ + ("[tool.pytest.ini_options]", "pytest 配置节"), + ("asyncio_mode", "异步测试模式"), + ("asyncio_default_fixture_loop_scope", "异步 fixture 作用域"), + ("[tool.coverage.run]", "覆盖率运行配置"), + ("[tool.coverage.report]", "覆盖率报告配置"), + ("fail_under = 80", "最低覆盖率要求 (80%)"), + ] + + for config_key, description in checks: + if config_key in content: + print_success(f"{description}: 已配置") + self.passed_checks += 1 + else: + print_error(f"{description}: 未配置") + self.failed_checks += 1 + + return True + + def check_dependencies(self) -> bool: + """检查测试依赖是否安装""" + print_header("3. 检查测试依赖") + + required_packages = [ + ("pytest", "Pytest 测试框架"), + ("pytest_asyncio", "pytest-asyncio 异步测试支持"), + ("pytest_cov", "pytest-cov 覆盖率插件"), + ("fastapi.testclient", "FastAPI TestClient"), + ("sqlalchemy", "SQLAlchemy ORM"), + ] + + for package_name, description in required_packages: + try: + spec = importlib.util.find_spec(package_name) + if spec is not None: + print_success(f"{description}: 已安装") + self.passed_checks += 1 + else: + print_error(f"{description}: 未安装") + self.failed_checks += 1 + except (ImportError, ModuleNotFoundError): + print_error(f"{description}: 未安装") + self.failed_checks += 1 + + return True + + def check_conftest(self) -> bool: + """检查 conftest.py 文件和 fixtures""" + print_header("4. 检查 conftest.py 和 fixtures") + + conftest = self.project_root / "tests" / "conftest.py" + if not self.check_file_exists(conftest, "conftest.py 文件"): + return False + + content = conftest.read_text() + + # 检查必需的 fixtures + required_fixtures = [ + ("test_db_file", "临时数据库文件 fixture"), + ("test_engine", "数据库引擎 fixture"), + ("db_session", "数据库会话 fixture"), + ("client", "FastAPI TestClient fixture"), + ("test_user", "测试用户 fixture"), + ("test_superuser", "超级用户 fixture"), + ("test_community", "测试社区 fixture"), + ("auth_headers", "认证头 fixture"), + ("superuser_auth_headers", "超级用户认证头 fixture"), + ("user_token", "用户 token fixture"), + ("superuser_token", "超级用户 token fixture"), + ] + + for fixture_name, description in required_fixtures: + if f"def {fixture_name}" in content: + print_success(f"{description}: 已定义") + self.passed_checks += 1 + else: + print_error(f"{description}: 未定义") + self.failed_checks += 1 + + return True + + def check_test_directory(self) -> bool: + """检查测试目录结构""" + print_header("5. 检查测试目录结构") + + tests_dir = self.project_root / "tests" + if not tests_dir.exists(): + print_error("tests 目录不存在") + self.failed_checks += 1 + return False + + print_success("tests 目录存在") + self.passed_checks += 1 + + # 检查测试文件 + test_files = list(tests_dir.glob("test_*.py")) + if test_files: + print_success(f"找到 {len(test_files)} 个测试文件") + self.passed_checks += 1 + + for test_file in test_files[:5]: # 只显示前 5 个 + print_info(f" - {test_file.name}") + + if len(test_files) > 5: + print_info(f" ... 还有 {len(test_files) - 5} 个测试文件") + else: + print_warning("没有找到测试文件 (test_*.py)") + self.warnings += 1 + + return True + + def check_pytest_markers(self) -> bool: + """检查 pytest 标记配置""" + print_header("6. 检查 pytest 标记配置") + + pytest_ini = self.project_root / "pytest.ini" + if not pytest_ini.exists(): + print_error("pytest.ini 不存在,无法检查标记") + self.failed_checks += 1 + return False + + content = pytest_ini.read_text() + + # 检查标记定义 + expected_markers = [ + "slow", + "integration", + "unit", + "api", + "auth", + "community", + "content", + "publish", + "analytics", + ] + + markers_section = "markers =" in content or "[markers]" in content + if not markers_section: + print_warning("未找到标记配置节") + self.warnings += 1 + return False + + for marker in expected_markers: + if marker in content: + print_success(f"标记 '{marker}': 已定义") + self.passed_checks += 1 + else: + print_warning(f"标记 '{marker}': 未定义(可选)") + self.warnings += 1 + + return True + + def check_coverage_config(self) -> bool: + """检查覆盖率配置""" + print_header("7. 检查覆盖率配置") + + pytest_ini = self.project_root / "pytest.ini" + if not pytest_ini.exists(): + print_error("pytest.ini 不存在") + self.failed_checks += 1 + return False + + content = pytest_ini.read_text() + + # 检查覆盖率配置 + checks = [ + ("[coverage:run]", "覆盖率运行配置节"), + ("source = app", "覆盖率源目录"), + ("omit =", "覆盖率排除配置"), + ("[coverage:report]", "覆盖率报告配置节"), + ("fail_under = 80", "最低覆盖率要求 (80%)"), + ("show_missing = True", "显示未覆盖行"), + ("[coverage:html]", "HTML 报告配置节"), + ("directory = htmlcov", "HTML 报告目录"), + ] + + for config_key, description in checks: + if config_key in content: + print_success(f"{description}: 已配置") + self.passed_checks += 1 + else: + print_error(f"{description}: 未配置") + self.failed_checks += 1 + + return True + + def check_asyncio_config(self) -> bool: + """检查 pytest-asyncio 配置""" + print_header("8. 检查 pytest-asyncio 配置") + + pytest_ini = self.project_root / "pytest.ini" + if not pytest_ini.exists(): + print_error("pytest.ini 不存在") + self.failed_checks += 1 + return False + + content = pytest_ini.read_text() + + # 检查 asyncio 配置 + if "asyncio_mode" in content: + if "asyncio_mode = auto" in content: + print_success("asyncio_mode: 设置为 auto (推荐)") + self.passed_checks += 1 + else: + print_warning("asyncio_mode: 已设置但不是 auto") + self.warnings += 1 + else: + print_error("asyncio_mode: 未配置") + self.failed_checks += 1 + + if "asyncio_default_fixture_loop_scope" in content: + if "asyncio_default_fixture_loop_scope = function" in content: + print_success("asyncio_default_fixture_loop_scope: 设置为 function") + self.passed_checks += 1 + else: + print_warning("asyncio_default_fixture_loop_scope: 已设置但不是 function") + self.warnings += 1 + else: + print_error("asyncio_default_fixture_loop_scope: 未配置") + self.failed_checks += 1 + + return True + + def check_requirements_dev(self) -> bool: + """检查 requirements-dev.txt""" + print_header("9. 检查 requirements-dev.txt") + + req_dev = self.project_root / "requirements-dev.txt" + if not self.check_file_exists(req_dev, "requirements-dev.txt 文件"): + return False + + content = req_dev.read_text() + + # 检查必需的依赖 + required_deps = [ + ("pytest", "Pytest"), + ("pytest-asyncio", "pytest-asyncio"), + ("pytest-cov", "pytest-cov"), + ] + + for dep_name, description in required_deps: + if dep_name in content: + print_success(f"{description}: 已列出") + self.passed_checks += 1 + else: + print_error(f"{description}: 未列出") + self.failed_checks += 1 + + return True + + def run_simple_test(self) -> bool: + """尝试运行一个简单的测试""" + print_header("10. 运行简单测试验证") + + try: + import subprocess + result = subprocess.run( + ["python", "-m", "pytest", "--version"], + capture_output=True, + text=True, + timeout=5, + ) + + if result.returncode == 0: + version = result.stdout.strip() + print_success(f"pytest 可以正常运行: {version}") + self.passed_checks += 1 + return True + else: + print_error("pytest 运行失败") + self.failed_checks += 1 + return False + except Exception as e: + print_error(f"无法运行 pytest: {e}") + self.failed_checks += 1 + return False + + def print_summary(self): + """打印总结""" + print_header("验证总结") + + total_checks = self.passed_checks + self.failed_checks + pass_rate = (self.passed_checks / total_checks * 100) if total_checks > 0 else 0 + + print(f"\n通过的检查: {Colors.GREEN}{self.passed_checks}{Colors.END}") + print(f"失败的检查: {Colors.RED}{self.failed_checks}{Colors.END}") + print(f"警告: {Colors.YELLOW}{self.warnings}{Colors.END}") + print(f"通过率: {Colors.BOLD}{pass_rate:.1f}%{Colors.END}\n") + + if self.failed_checks == 0: + print(f"{Colors.GREEN}{Colors.BOLD}✓ 所有配置检查通过!{Colors.END}") + print(f"\n{Colors.BLUE}可以开始运行测试了:{Colors.END}") + print(f" cd backend") + print(f" pytest") + return 0 + else: + print(f"{Colors.RED}{Colors.BOLD}✗ 有 {self.failed_checks} 个配置问题需要解决{Colors.END}") + print(f"\n{Colors.BLUE}请检查上述失败的项目并修复配置。{Colors.END}") + return 1 + + def run_all_checks(self) -> int: + """运行所有检查""" + print(f"{Colors.BOLD}{'=' * 70}{Colors.END}") + print(f"{Colors.BOLD}Pytest 测试配置验证{Colors.END}") + print(f"{Colors.BOLD}{'=' * 70}{Colors.END}") + + # 运行所有检查 + self.check_pytest_ini() + self.check_pyproject_toml() + self.check_dependencies() + self.check_conftest() + self.check_test_directory() + self.check_pytest_markers() + self.check_coverage_config() + self.check_asyncio_config() + self.check_requirements_dev() + self.run_simple_test() + + # 打印总结 + return self.print_summary() + + +def main(): + """主函数""" + validator = TestConfigValidator() + exit_code = validator.run_all_checks() + sys.exit(exit_code) + + +if __name__ == "__main__": + main()