python-testing

# Python 测试模式（Python Testing Patterns） 使用 pytest、测试驱动开发（TDD）方法论及最佳实践的 Python 应用程序全面测试策略。 ## 何时激活 - 编写新的 Python 代码时（遵循 TDD：红、绿、重构） - 为 Python 项目设计测试套件时 - 审查 Python 测试覆盖率时 - 搭建测试基础设施时 ## 核心测试理念 ### 测试驱动开发（TDD） 始终遵循 TDD 循环： 1. **红（RED）**：为期望的行为编写一个失败的测试 2. **绿（GREEN）**：编写最少量的代码使测试通过 3. **重构（REFACTOR）**：在保持测试通过的前提下优化代码 ```python # 步骤 1：编写失败的测试 (RED) def test_add_numbers(): result = add(2, 3) assert result == 5 # 步骤 2：编写最小实现 (GREEN) def add(a, b): return a + b # 步骤 3：根据需要进行重构 (REFACTOR) ``` ### 覆盖率要求 - **目标**：80% 以上的代码覆盖率 - **关键路径**：必须达到 100% 覆盖率 - 使用 `pytest --cov` 来衡量覆盖率 ```bash pytest --cov=mypackage --cov-report=term-missing --cov-report=html ``` ## pytest 基础 ### 基本测试结构 ```python import pytest def test_addition(): """测试基础加法。""" assert 2 + 2 == 4 def test_string_uppercase(): """测试字符串大写转换。""" text = "hello" assert text.upper() == "HELLO" def test_list_append(): """测试列表追加。""" items = [1, 2, 3] items.append(4) assert 4 in items assert len(items) == 4 ``` ### 断言（Assertions） ```python # 相等性 assert result == expected # 不等性 assert result != unexpected # 真值 assert result # Truthy assert not result # Falsy assert result is True # 精确为 True assert result is False # 精确为 False assert result is None # 精确为 None # 成员资格 assert item in collection assert item not in collection # 比较 assert result > 0 assert 0 <= result <= 100 # 类型检查 assert isinstance(result, str) # 异常测试（推荐做法） with pytest.raises(ValueError): raise ValueError("error message") # 检查异常消息 with pytest.raises(ValueError, match="invalid input"): raise ValueError("invalid input provided") # 检查异常属性 with pytest.raises(ValueError) as exc_info: raise ValueError("error message") assert str(exc_info.value) == "error message" ``` ## Fixtures ### 基础 Fixture 用法 ```python import pytest @pytest.fixture def sample_data(): """提供示例数据的 Fixture。""" return {"name": "Alice", "age": 30} def test_sample_data(sample_data): """使用 fixture 的测试。""" assert sample_data["name"] == "Alice" assert sample_data["age"] == 30 ``` ### 带有设置（Setup）和清理（Teardown）的 Fixture ```python @pytest.fixture def database(): """带有设置和清理逻辑的 Fixture。""" # 设置 (Setup) db = Database(":memory:") db.create_tables() db.insert_test_data() yield db # 提供给测试使用 # 清理 (Teardown) db.close() def test_database_query(database): """测试数据库操作。""" result = database.query("SELECT * FROM users") assert len(result) > 0 ``` ### Fixture 作用域（Scopes） ```python # 函数级作用域 (默认) - 每个测试运行一次 @pytest.fixture def temp_file(): with open("temp.txt", "w") as f: yield f os.remove("temp.txt") # 模块级作用域 - 每个模块运行一次 @pytest.fixture(scope="module") def module_db(): db = Database(":memory:") db.create_tables() yield db db.close() # 会话级作用域 - 整个测试会话运行一次 @pytest.fixture(scope="session") def shared_resource(): resource = ExpensiveResource() yield resource resource.cleanup() ``` ### 带参数的 Fixture ```python @pytest.fixture(params=[1, 2, 3]) def number(request): """参数化 Fixture。""" return request.param def test_numbers(number): """测试将运行 3 次，每个参数一次。""" assert number > 0 ``` ### 使用多个 Fixture ```python @pytest.fixture def user(): return User(id=1, name="Alice") @pytest.fixture def admin(): return User(id=2, name="Admin", role="admin") def test_user_admin_interaction(user, admin): """同时使用多个 fixture 的测试。""" assert admin.can_manage(user) ``` ### 自动使用（Autouse） Fixture ```python @pytest.fixture(autouse=True) def reset_config(): """在每个测试前自动运行。""" Config.reset() yield Config.cleanup() def test_without_fixture_call(): # reset_config 会自动运行 assert Config.get_setting("debug") is False ``` ### 用于共享 Fixture 的 conftest.py ```python # tests/conftest.py import pytest @pytest.fixture def client(): """供所有测试共享的 fixture。""" app = create_app(testing=True) with app.test_client() as client: yield client @pytest.fixture def auth_headers(client): """为 API 测试生成认证头。""" response = client.post("/api/login", json={ "username": "test", "password": "test" }) token = response.json["token"] return {"Authorization": f"Bearer {token}"} ``` ## 参数化（Parametrization） ### 基础参数化 ```python @pytest.mark.parametrize("input,expected", [ ("hello", "HELLO"), ("world", "WORLD"), ("PyThOn", "PYTHON"), ]) def test_uppercase(input, expected): """测试将使用不同的输入运行 3 次。""" assert input.upper() == expected ``` ### 多个参数 ```python @pytest.mark.parametrize("a,b,expected", [ (2, 3, 5), (0, 0, 0), (-1, 1, 0), (100, 200, 300), ]) def test_add(a, b, expected): """使用多组输入测试加法。""" assert add(a, b) == expected ``` ### 带 ID 的参数化 ```python @pytest.mark.parametrize("input,expected", [ ("valid@email.com", True), ("invalid", False), ("@no-domain.com", False), ], ids=["valid-email", "missing-at", "missing-domain"]) def test_email_validation(input, expected): """通过可读的测试 ID 测试电子邮件验证。""" assert is_valid_email(input) is expected ``` ### 参数化 Fixtures ```python @pytest.fixture(params=["sqlite", "postgresql", "mysql"]) def db(request): """针对多个数据库后端进行测试。""" if request.param == "sqlite": return Database(":memory:") elif request.param == "postgresql": return Database("postgresql://localhost/test") elif request.param == "mysql": return Database("mysql://localhost/test") def test_database_operations(db): """测试将运行 3 次，每个数据库一次。""" result = db.query("SELECT 1") assert result is not None ``` ## 标记（Markers）与测试选择 ### 自定义标记 ```python # 标记慢速测试 @pytest.mark.slow def test_slow_operation(): time.sleep(5) # 标记集成测试 @pytest.mark.integration def test_api_integration(): response = requests.get("https://api.example.com") assert response.status_code == 200 # 标记单元测试 @pytest.mark.unit def test_unit_logic(): assert calculate(2, 3) == 5 ``` ### 运行特定测试 ```bash # 仅运行非慢速测试 pytest -m "not slow" # 仅运行集成测试 pytest -m integration # 运行集成测试或慢速测试 pytest -m "integration or slow" # 运行标记为单元测试且非慢速的测试 pytest -m "unit and not slow" ``` ### 在 pytest.ini 中配置标记 ```ini [pytest] markers = slow: 将测试标记为慢速 integration: 将测试标记为集成测试 unit: 将测试标记为单元测试 django: 将测试标记为需要 Django 环境 ``` ## Mocking 与 Patching ### Mock 函数 ```python from unittest.mock import patch, Mock @patch("mypackage.external_api_call") def test_with_mock(api_call_mock): """使用 mock 的外部 API 进行测试。""" api_call_mock.return_value = {"status": "success"} result = my_function() api_call_mock.assert_called_once() assert result["status"] == "success" ``` ### Mock 返回值 ```python @patch("mypackage.Database.connect") def test_database_connection(connect_mock): """使用 mock 的数据库连接进行测试。""" connect_mock.return_value = MockConnection() db = Database() db.connect() connect_mock.assert_called_once_with("localhost") ``` ### Mock 异常 ```python @patch("mypackage.api_call") def test_api_error_handling(api_call_mock): """使用 mock 异常测试错误处理。""" api_call_mock.side_effect = ConnectionError("Network error") with pytest.raises(ConnectionError): api_call() api_call_mock.assert_called_once() ``` ### Mock 上下文管理器（Context Managers） ```python @patch("builtins.open", new_callable=mock_open) def test_file_reading(mock_file): """使用 mock 的 open 测试文件读取。""" mock_file.return_value.read.return_value = "file content" result = read_file("test.txt") mock_file.assert_called_once_with("test.txt", "r") assert result == "file content" ``` ### 使用 Autospec ```python @patch("mypackage.DBConnection", autospec=True) def test_autospec(db_mock): """使用 autospec 捕获 API 滥用。""" db = db_mock.return_value db.query("SELECT * FROM users") # 如果 DBConnection 没有 query 方法，此处将失败 db_mock.assert_called_once() ``` ### Mock 类实例 ```python class TestUserService: @patch("mypackage.UserRepository") def test_create_user(self, repo_mock): """使用 mock 的仓库进行用户创建测试。""" repo_mock.return_value.save.return_value = User(id=1, name="Alice") service = UserService(repo_mock.return_value) user = service.create_user(name="Alice") assert user.name == "Alice" repo_mock.return_value.save.assert_called_once() ``` ### Mock 属性（Property） ```python @pytest.fixture def mock_config(): """创建一个带有属性的 mock。""" config = Mock() type(config).debug = PropertyMock(return_value=True) type(config).api_key = PropertyMock(return_value="test-key") return config def test_with_mock_config(mock_config): """使用 mock 配置属性进行测试。""" assert mock_config.debug is True assert mock_config.api_key == "test-key" ``` ## 测试异步代码 ### 使用 pytest-asyncio 进行异步测试 ```python import pytest @pytest.mark.asyncio async def test_async_function(): """测试异步函数。""" result = await async_add(2, 3) assert result == 5 @pytest.mark.asyncio async def test_async_with_fixture(async_client): """在异步 fixture 下进行异步测试。""" response = await async_client.get("/api/users")