测试策略设计：单元/集成/E2E分层策略

Python 测试与调试专题 · 构建完整且高效的测试体系

专题：Python 测试与调试系统学习

章节：十、CI/CD与工程实践篇

关键词：Python, 测试, 调试, 测试策略, 测试金字塔, 测试分层, 测试左移, 测试设计, 项目测试

一、测试策略概述

测试策略（Test Strategy）是软件质量保障体系的顶层设计，它定义了"测什么、怎么测、何时测、测多少"四个核心问题的答案。在Python项目中，一个良好的测试策略能够帮助团队在有限的资源和时间约束下，最大化测试的投资回报率（ROI）。测试策略不同于具体的测试计划，它更关注方法论层面的选择，而非某个具体迭代的测试执行细节。

测试金字塔（Test Pyramid）由Mike Cohn在2009年提出，是测试策略设计中最经典的理论模型。金字塔从下到上分为三层：单元测试（Unit Tests）、集成测试（Integration Tests）、端到端测试（E2E Tests）。底层测试数量多、速度快、成本低；顶层测试数量少、速度慢、成本高。这个模型强调：应该投入最多精力在单元测试上，其次是集成测试，最少的是E2E测试。比例大致遵循70/20/10法则，即70%单元测试、20%集成测试、10%E2E测试。

与测试金字塔相对的是"冰激凌锥"反模式（Ice-Cream Cone Anti-Pattern）。这种模式表现为：E2E测试数量过多、单元测试严重不足。常见于过度依赖UI自动化测试的团队。冰激凌锥的后果包括：测试运行时间过长（一套完整的E2E测试可能运行数小时）、测试不稳定（UI元素稍变即红）、反馈周期太长（开发者提交代码后需要等待很久才能知道测试结果）、维护成本高昂（定位失败原因困难）。避免冰激凌锥的关键在于始终坚持"尽可能在下层发现缺陷"的原则。

除了测试金字塔，测试象限（Agile Testing Quadrants）是另一个重要的策略框架。由Brian Marick提出，它将测试分为四个象限：Q1面向技术的支持团队测试（单元测试、组件测试）、Q2面向业务的支持团队测试（功能测试、用户故事测试）、Q3面向业务的评价产品测试（探索性测试、用户验收测试）、Q4面向技术的评价产品测试（性能测试、安全测试）。测试策略需要覆盖所有四个象限，但各象限的投入比例需根据项目特点动态调整。

核心原则：测试策略设计应遵循"尽早测试（Shift Left）"原则，将测试活动尽可能向开发周期的早期移动。缺陷发现得越早，修复成本越低。研究表明，在需求阶段发现一个缺陷的成本如果是1，那么在编码阶段就是10，在测试阶段是50，在生产环境则是500甚至更高。因此，单元测试和静态分析等左移实践是提升整体质量效率的关键。

策略平衡考量要素

设计合理的测试策略需要权衡以下多维因素：项目类型（Web应用、数据处理、嵌入式系统各有不同）、团队规模（小团队适合简洁策略，大团队需要更严格的分层）、技术栈（动态语言如Python需要更多类型检查和契约测试）、发布频率（每日发布需要更强大的自动化回归能力）、合规要求（金融医疗领域需要更全面的测试覆盖）。常见的策略平衡方式包括：核心模块采用高覆盖率要求（90%+），边缘模块适当放宽（60%）甚至不写单元测试，仅通过集成测试覆盖。

# 测试策略配置文件示例：pytest.ini
[pytest]
# 标记定义用于测试分层
markers =
    unit: 单元测试，快速、隔离、无外部依赖
    integration: 集成测试，依赖数据库或外部服务
    e2e: 端到端测试，模拟真实用户操作
    slow: 慢速测试，需要长时间运行
    smoke: 冒烟测试，核心功能快速验证

# 默认排除慢速和E2E测试
addopts =
    -m "not slow and not e2e"
    --strict-markers
    --tb=short
    -q

# 覆盖率配置
testpaths = tests
python_files = test_*.py

# 运行不同层次测试的命令
# 仅运行单元测试
pytest -m unit -v

# 运行单元+集成测试
pytest -m "not e2e" -v

# 运行全部测试（包括E2E）
pytest -v --run-e2e

# 运行冒烟测试
pytest -m smoke -v

# 生成覆盖率报告
pytest --cov=src --cov-report=html --cov-report=term -m "not e2e"

# conftest.py - 添加自定义命令行选项控制E2E测试
import pytest

def pytest_addoption(parser):
    parser.addoption(
        "--run-e2e",
        action="store_true",
        default=False,
        help="运行端到端测试（默认跳过）"
    )

def pytest_configure(config):
    config.addinivalue_line(
        "markers",
        "e2e: 标记为端到端测试，需要 --run-e2e 选项"
    )

def pytest_collection_modifyitems(config, items):
    if not config.getoption("--run-e2e"):
        skip_e2e = pytest.mark.skip(reason="需要 --run-e2e 选项来运行")
        for item in items:
            if "e2e" in item.keywords:
                item.add_marker(skip_e2e)

二、单元测试策略

单元测试（Unit Testing）是测试金字塔的基石，也是投入产出比最高的测试层次。单元测试的目标是验证代码中最小可测试单元（通常是函数或方法）的行为是否符合预期。在Python项目中，单元测试的主要框架是unittest（标准库）和pytest（第三方）。一个成熟的单元测试策略需要关注测试原则、Mock策略、覆盖率目标和测试方法论等多个维度。

FIRST原则

优秀的单元测试应当遵循FIRST原则：Fast（快速）——单个单元测试应在毫秒级完成，整个单元测试套件应在数秒内运行完毕；Isolated（隔离）——测试之间互不依赖，每个测试可以独立运行，且运行顺序不影响结果；Repeatable（可重复）——在任何环境、任何时间运行同一测试，结果都应当一致；Self-validating（自验证）——测试结果应为布尔值（通过/失败），无需人工检查；Timely（及时）——测试应该在实际编码之前（TDD）或之后立即编写，而非等到项目后期集中补写。

Mock使用原则

Mock是单元测试隔离外部依赖的核心手段。良好Mock策略的要点包括：只Mock外部依赖（如网络请求、数据库、文件系统），不Mock被测系统内部的间接依赖；使用autospec确保Mock对象与被Mock对象的接口保持一致，防止生产接口变更后Mock仍然通过；优先使用fixture创建共享Mock对象，避免在每个测试中重复配置；不要过度Mock，如果Mock链路过长，说明被测对象的设计可能存在问题（需要重构）；对于简单对象，考虑使用桩（Stub）而非Mock，减少测试的脆弱性。

测试覆盖率目标

测试覆盖率是衡量单元测试充分性的重要指标，但不应是唯一指标。常见的覆盖率目标设定策略：核心业务逻辑模块——行覆盖率达到90%以上，分支覆盖率85%以上；工具类和辅助模块——行覆盖率70-80%；配置类和数据模型——覆盖率不做硬性要求，但基本构造和关键方法需要覆盖。需要注意的是，100%覆盖率并不代表代码没有缺陷。更重要的是测试的"有效性"——是否覆盖了关键的业务逻辑和边界条件。在CI流程中，可以设定覆盖率门禁：新增代码的覆盖率不得低于80%，整体项目覆盖率不得低于75%，否则CI构建失败。

# 遵循FIRST原则的单元测试示例
import pytest
from datetime import datetime, timedelta
from src.models import Order
from src.services import DiscountCalculator

class TestDiscountCalculator:
    """打折计算器的单元测试——符合FIRST原则"""

    def test_calculate_basic_discount(self):
        """基本打折计算：满100减10（Fast + Self-validating）"""
        calc = DiscountCalculator()
        result = calc.apply_discount(total=100.0, rate=0.1)
        assert result == 90.0
        assert isinstance(result, float)

    def test_zero_discount_returns_original_price(self):
        """零折扣返回原价（Isolated + Repeatable）"""
        calc = DiscountCalculator()
        assert calc.apply_discount(200.0, 0.0) == 200.0

    def test_full_discount_returns_zero(self):
        """100%折扣返回0"""
        calc = DiscountCalculator()
        assert calc.apply_discount(200.0, 1.0) == 0.0

    def test_invalid_rate_raises_error(self):
        """无效折扣率抛出异常"""
        calc = DiscountCalculator()
        with pytest.raises(ValueError, match="折扣率必须在0到1之间"):
            calc.apply_discount(100.0, 1.5)

    @pytest.mark.parametrize("total,rate,expected", [
        (100.0, 0.1, 90.0),
        (0.0, 0.1, 0.0),
        (99.99, 0.2, 79.992),
        (1000.0, 0.5, 500.0),
    ])
    def test_varied_discount_scenarios(self, total, rate, expected):
        """参数化测试多种折扣场景"""
        calc = DiscountCalculator()
        result = calc.apply_discount(total, rate)
        assert abs(result - expected) < 0.001  # 浮点数比较

# Mock外部依赖的单元测试
from unittest.mock import Mock, patch
import pytest
from src.services import PaymentService
from src.gateways import PaymentGateway

class TestPaymentService:
    """支付服务测试——Mock外部网关"""

    @pytest.fixture
    def mock_gateway(self):
        """创建Mock支付网关"""
        gateway = Mock(spec=PaymentGateway)
        gateway.charge.return_value = {"status": "success", "txn_id": "TXN123"}
        gateway.refund.return_value = {"status": "success"}
        return gateway

    def test_successful_payment(self, mock_gateway):
        """测试支付成功流程"""
        service = PaymentService(gateway=mock_gateway)
        result = service.process_payment(
            user_id=1,
            amount=99.99,
            currency="USD"
        )
        assert result["status"] == "success"
        assert result["txn_id"] == "TXN123"
        mock_gateway.charge.assert_called_once_with(
            amount=99.99,
            currency="USD",
            source="default"
        )

    def test_payment_with_coupon(self, mock_gateway):
        """测试带优惠券的支付"""
        mock_gateway.charge.return_value = {
            "status": "success",
            "txn_id": "TXN456",
            "discount_applied": 10.0
        }
        service = PaymentService(gateway=mock_gateway)
        result = service.process_payment(
            user_id=1, amount=99.99,
            currency="USD", coupon_code="SAVE10"
        )
        assert result["discount_applied"] == 10.0
        # 验证调用参数中包含优惠券信息
        call_kwargs = mock_gateway.charge.call_args[1]
        assert call_kwargs.get("coupon") == "SAVE10"

    def test_payment_gateway_error(self, mock_gateway):
        """测试网关错误处理"""
        mock_gateway.charge.side_effect = ConnectionError("网关超时")
        service = PaymentService(gateway=mock_gateway)
        with pytest.raises(ConnectionError):
            service.process_payment(user_id=1, amount=99.99, currency="USD")

    def test_autospec_prevents_interface_mismatch(self):
        """autospec确保Mock与被Mock对象接口一致"""
        gateway = Mock(spec=PaymentGateway)
        # 如果调用不存在的方法，会失败
        with pytest.raises(AttributeError):
            gateway.nonexistent_method()

# 测试覆盖率配置：.coveragerc
[run]
source = src
omit =
    */migrations/*
    */tests/*
    */setup.py
    */__init__.py

[report]
# 覆盖率门禁
fail_under = 75
# 排除模板和调试代码
exclude_lines =
    pragma: no cover
    def __repr__
    if __name__ == .__main__.:
    raise NotImplementedError
    raise AssertionError
    pass

[html]
directory = coverage_html_report

三、集成测试策略

集成测试（Integration Testing）位于测试金字塔的中间层，验证不同模块或系统之间的交互是否正确。在Python项目中，集成测试主要关注：数据库操作（CRUD、迁移、事务）、外部服务调用（REST API、消息队列、文件存储）、模块间接口（服务层与数据层交互、事件驱动通信）。集成测试的策略设计需要在"真实性与速度"之间取得平衡——使用真实的数据库实例而非Mock，但可以通过内存数据库（如SQLite :memory:）或测试容器（Testcontainers）来提升运行速度。

数据库测试策略

数据库集成测试的策略选择取决于项目复杂度：小型项目可以采用SQLite内存模式，每次测试前创建完整Schema并插入种子数据，测试结束后自动回滚事务；中型项目推荐使用Testcontainers启动PostgreSQL/MySQL的Docker容器，兼顾真实性与开发环境一致性；大型项目或需要验证特定数据库特性的场景，应连接专用测试数据库（但不能是生产库），并配合数据库迁移工具（Alembic）自动维护Schema版本。无论采用哪种方案，关键原则包括：每个测试独立事务、测试前后清理数据、使用工厂类创建测试数据而非硬编码SQL。

外部服务测试策略

对于依赖外部HTTP服务（支付网关、短信平台、第三方API）的集成测试，有三种主流策略：一是使用responses/httpx-mock库拦截HTTP请求并返回预设响应，适用于测试本系统的请求构建和响应处理逻辑；二是使用WireMock搭建独立的模拟HTTP服务器，适用于需要验证复杂请求匹配和响应模板的场景；三是使用vcrpy录制真实HTTP交互并回放，适用于第三方API文档不完善的场景。选择策略的依据是：需要测试本系统（选择策略一）、本系统与外部系统的契约（选择策略二）、外部系统的行为（选择策略三）。

契约测试的位置

契约测试（Contract Testing）是集成测试的重要补充。在微服务架构中，契约测试用于验证服务间API协议的兼容性。常见的实践是：消费者端编写契约（描述期望的请求和响应），将契约发布到Pact Broker，提供者端拉取契约并验证自己的实现是否满足所有消费者的期望。契约测试的位置在单元测试之上、端到端测试之下，它们比集成测试更轻量（只验证接口协议，不验证业务逻辑），但比单元测试更"真实"（基于实际的HTTP请求/响应格式）。在服务数量超过5个的微服务项目中，契约测试是维护服务间兼容性的最低成本方案。

# 数据库集成测试：使用SQLite内存模式
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from src.models import Base, User, Order

@pytest.fixture
def db_session():
    """创建内存SQLite数据库会话，每个测试独立"""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    try:
        yield session
    finally:
        session.close()
        Base.metadata.drop_all(engine)

class TestUserRepository:
    """用户仓库集成测试"""

    def test_create_user(self, db_session):
        """测试用户创建和持久化"""
        user = User(
            username="test_user",
            email="test@example.com",
            is_active=True
        )
        db_session.add(user)
        db_session.commit()

        saved_user = db_session.query(User).filter_by(
            username="test_user"
        ).first()
        assert saved_user is not None
        assert saved_user.email == "test@example.com"
        assert saved_user.is_active is True
        assert saved_user.created_at is not None

    def test_unique_username_constraint(self, db_session):
        """测试用户名唯一约束"""
        user1 = User(username="same_name", email="a@test.com")
        user2 = User(username="same_name", email="b@test.com")
        db_session.add(user1)
        db_session.commit()

        db_session.add(user2)
        with pytest.raises(Exception) as excinfo:
            db_session.commit()
        assert "UNIQUE" in str(excinfo.value) or "unique" in str(excinfo.value).lower()

    def test_user_with_orders_relationship(self, db_session):
        """测试用户与订单的关联关系"""
        user = User(username="buyer", email="buyer@test.com")
        db_session.add(user)
        db_session.flush()

        order = Order(
            user_id=user.id,
            total_amount=199.99,
            status="pending"
        )
        db_session.add(order)
        db_session.commit()

        # 验证关联加载
        assert len(user.orders) == 1
        assert user.orders[0].status == "pending"

# 外部服务集成测试：使用pytest-httpx
import pytest
from httpx import Response
from src.services import WeatherService

@pytest.mark.integration
class TestWeatherService:
    """天气服务集成测试——模拟HTTP响应"""

    @pytest.fixture
    def service(self):
        return WeatherService(api_key="test-key")

    def test_get_temperature_success(self, service, httpx_mock):
        """测试成功获取温度"""
        httpx_mock.add_response(
            url="https://api.weather.com/v1/current?city=Beijing",
            json={
                "city": "Beijing",
                "temperature": 22.5,
                "unit": "celsius",
                "humidity": 65
            },
            status_code=200
        )
        result = service.get_temperature("Beijing")
        assert result == 22.5

    def test_city_not_found(self, service, httpx_mock):
        """测试城市不存在的情况"""
        httpx_mock.add_response(
            url="https://api.weather.com/v1/current?city=Nonexistent",
            json={"error": "City not found"},
            status_code=404
        )
        with pytest.raises(ValueError, match="找不到城市信息"):
            service.get_temperature("Nonexistent")

    def test_api_timeout(self, service, httpx_mock):
        """测试API超时处理"""
        httpx_mock.add_exception(
            TimeoutException("请求超时"),
            url="https://api.weather.com/v1/current?city=Beijing"
        )
        result = service.get_temperature("Beijing")
        # 带有降级策略：超时时返回缓存数据或默认值
        assert result == service.fallback_temperature

# 使用Testcontainers集成测试PostgreSQL
import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine, text
from src.models import Base

@pytest.mark.integration
class TestPostgresIntegration:
    """使用Testcontainers启动真实PostgreSQL的集成测试"""

    @classmethod
    def setup_class(cls):
        cls.postgres = PostgresContainer("postgres:14-alpine")
        cls.postgres.start()
        cls.engine = create_engine(cls.postgres.get_connection_url())
        Base.metadata.create_all(cls.engine)

    @classmethod
    def teardown_class(cls):
        cls.engine.dispose()
        cls.postgres.stop()

    def test_jsonb_field_support(self):
        """测试PostgreSQL特有JSONB字段"""
        with self.engine.connect() as conn:
            conn.execute(text("""
                CREATE TABLE test_jsonb (
                    id SERIAL PRIMARY KEY,
                    data JSONB
                )
            """))
            conn.execute(
                text("INSERT INTO test_jsonb (data) VALUES (:data)"),
                {"data": '{"key": "value", "nested": {"a": 1}}'}
            )
            conn.commit()
            result = conn.execute(
                text("SELECT data->>'key' as key_val FROM test_jsonb")
            ).fetchone()
            assert result[0] == "value"

四、E2E测试策略

端到端测试（End-to-End Testing）位于测试金字塔的顶层，验证整个系统从用户界面到后端数据库的业务流程是否完整正确。E2E测试模拟真实用户操作，覆盖系统中最关键的用户路径。在Python生态中，常用的E2E测试工具包括Selenium（浏览器自动化）、Playwright（微软出品，支持多浏览器）、Cypress（前端友好）等。E2E测试的策略核心是"精选关键路径、控制测试数量、确保测试稳定"。

E2E测试范围选择

E2E测试不应试图覆盖所有功能，这是常见的错误。正确的做法是：只覆盖最核心的"快乐路径"（Happy Path）和部分关键异常路径。选择标准包括：用户使用频率最高（如登录、搜索、下单）、业务价值最大（如支付结算、数据导出）、涉及系统最多（需要多个服务协同工作的场景）。对于非核心功能，应通过低层级的集成测试或单元测试来覆盖。一个经验法则是：如果某个E2E测试失败后，可通过单元测试或集成测试快速定位问题，那么该场景更适合用下层测试覆盖。

Page Object设计模式

Page Object模式是E2E测试中最常见的设计模式，它将页面封装为Python类，页面上的元素定位和操作方法封装为类的方法。这样做的好处包括：分离测试逻辑和页面实现细节（页面UI变化只需要修改Page Object，不影响测试用例）、提高代码复用性（多个测试可以共享同一个Page Object）、增强可维护性（定位器集中在Page Object中）。实践中，通常还会引入Page Component模式，将页面中的重复组件（如导航栏、搜索框、分页组件）提取为独立的组件类，进一步减少重复代码。

E2E测试稳定性

E2E测试的不稳定性（Flaky Tests）是实践中最大的挑战。提升稳定性的常见策略包括：使用显式等待而非固定sleep：等待特定元素出现、可点击、文本匹配等条件，而非盲目等待固定时间；测试隔离：每个E2E测试应有独立的测试数据，避免测试间相互影响；重试机制：对于已知的不稳定因素（如网络延迟、渲染时序），可以配置自动重试（如pytest-rerunfailures）；幂等性设计：确保重复运行同一测试产生相同结果；环境独立：E2E测试应在独立的测试环境而非开发或预发布环境运行。

# Page Object模式实现E2E测试
from playwright.sync_api import Page

class LoginPage:
    """登录页面Page Object"""

    def __init__(self, page: Page):
        self.page = page
        self.username_input = page.locator("#username")
        self.password_input = page.locator("#password")
        self.login_button = page.locator("button[type='submit']")
        self.error_message = page.locator(".error-message")
        self.remember_checkbox = page.locator("#remember-me")

    def navigate(self):
        """导航到登录页面"""
        self.page.goto("https://test.example.com/login")
        return self

    def login(self, username: str, password: str):
        """执行登录操作"""
        self.username_input.fill(username)
        self.password_input.fill(password)
        self.login_button.click()
        return self

    def login_with_remember(self, username: str, password: str):
        """勾选记住我后登录"""
        self.remember_checkbox.check()
        return self.login(username, password)

    def get_error_message(self) -> str:
        """获取登录错误信息"""
        self.error_message.wait_for(state="visible")
        return self.error_message.text_content()

    def is_login_successful(self) -> bool:
        """验证登录是否成功（页面跳转到首页）"""
        return self.page.url.endswith("/dashboard")


class DashboardPage:
    """仪表盘页面Page Object"""

    def __init__(self, page: Page):
        self.page = page
        self.welcome_message = page.locator(".welcome-message")
        self.user_avatar = page.locator(".user-avatar")
        self.logout_button = page.locator("#logout-btn")

    def get_welcome_text(self) -> str:
        """获取欢迎文本"""
        self.welcome_message.wait_for(state="visible")
        return self.welcome_message.text_content()

    def logout(self):
        """执行退出登录"""
        self.logout_button.click()
        return LoginPage(self.page)

# E2E测试用例：用户完整的购物流程
import pytest
from playwright.sync_api import expect

class TestShoppingFlow:
    """电商系统核心购物流程E2E测试"""

    @pytest.fixture(autouse=True)
    def setup(self, page, test_user):
        """每个测试前登录系统"""
        login_page = LoginPage(page)
        login_page.navigate().login(
            username=test_user["username"],
            password=test_user["password"]
        )
        # 等待登录完成
        expect(page).to_have_url("**/dashboard")

    def test_complete_purchase_flow(self, page, test_product):
        """完整购买流程：搜索→加入购物车→结算→支付→确认"""
        # 1. 搜索商品
        search_page = SearchPage(page)
        search_page.search(test_product["name"])
        search_page.select_first_result()
        expect(page.locator(".product-title")).to_contain_text(
            test_product["name"]
        )

        # 2. 加入购物车
        product_page = ProductDetailPage(page)
        product_page.select_quantity(2)
        product_page.add_to_cart()
        expect(page.locator(".cart-badge")).to_contain_text("2")

        # 3. 去结算
        cart_page = CartPage(page)
        cart_page.navigate()
        expect(cart_page.get_total_amount()).to_be_greater_than(0)
        cart_page.proceed_to_checkout()

        # 4. 填写配送信息
        checkout_page = CheckoutPage(page)
        checkout_page.fill_shipping_address(
            name="张三",
            phone="13800138000",
            address="北京市海淀区中关村大街1号",
        )
        checkout_page.select_shipping_method("标准配送")

        # 5. 选择支付方式并支付
        checkout_page.select_payment("支付宝")
        payment_page = PaymentPage(page)
        payment_page.complete_payment()

        # 6. 确认订单成功
        expect(page.locator(".order-success")).to_be_visible()
        expect(page.locator(".order-number")).not_to_be_empty()

# E2E测试配置与稳定性处理
import pytest

@pytest.mark.e2e
class TestUserManagementE2E:
    """用户管理的E2E测试——包含稳定性策略"""

    @pytest.mark.flaky(reruns=2, reruns_delay=5)
    def test_user_registration_flow(self, page, clean_database):
        """用户注册流程（含重试机制）"""
        # 使用显式等待替代隐式等待
        page.goto("https://test.example.com/register")

        # 等待表单完全加载
        page.wait_for_selector("#registration-form", state="visible")
        page.wait_for_selector("#username", state="attached")

        # 使用唯一用户名避免数据冲突
        import uuid
        unique_username = f"test_user_{uuid.uuid4().hex[:8]}"

        page.fill("#username", unique_username)
        page.fill("#email", f"{unique_username}@test.com")
        page.fill("#password", "TestPass123!")
        page.fill("#confirm-password", "TestPass123!")

        # 勾选用户协议
        page.check("#agree-terms")

        # 提交注册表单
        page.click("button[type='submit']")

        # 等待跳转到验证页面或欢迎页面
        page.wait_for_url("**/welcome**", timeout=10000)
        assert page.locator(".welcome-title").is_visible()

    def test_password_reset_flow(self, page):
        """密码重置流程"""
        page.goto("https://test.example.com/login")

        # 点击"忘记密码"
        page.click("text=忘记密码")

        # 等待密码重置页面
        page.wait_for_selector("#reset-password-form", state="visible")

        # 输入注册邮箱
        page.fill("#email", "persistent_user@test.com")
        page.click("button[type='submit']")

        # 等待发送成功提示
        page.wait_for_selector(".success-message", state="visible")
        assert page.locator(".success-message").text_content().strip() != ""

五、测试比例与ROI

测试比例（Test Ratio）决定了测试资源的分配方案，直接影响到测试的投资回报率（Return on Investment, ROI）。最经典的分配法则是70/20/10法则：70%的测试为单元测试，20%为集成测试，10%为E2E测试。这个比例并非偶然，而是基于各层测试的成本效益分析得出的经验值。在实际项目中，该比例需要根据项目特点动态调整：金融系统可能需要增加集成测试的比例（30%），而CMS系统可以适当降低E2E比例（5%）。

各层测试ROI分析

单元测试的ROI最高，原因如下：编写成本低（平均每个测试5-15分钟）、执行速度快（毫秒级）、定位精准（直接定位到函数和行号）、维护成本低（与被测试代码紧耦合，重构时需要更新但改动范围有限）。集成测试的ROI中等：编写成本较高（需要搭建测试环境）、执行速度中等（秒级到分钟级）、定位相对精准（可以定位到接口级别）。E2E测试的ROI最低：编写成本高（需要复杂的Page Object和测试数据准备）、执行速度慢（分钟到小时级）、定位困难（失败可能由前端、后端、网络、数据库任意一层引起）、最不稳定（Flaky率通常在5-15%）。

维护成本与执行时间

测试的维护成本随层次上升而指数级增加：一个单元测试的生命周期成本约为50-100元，集成测试约为300-500元，E2E测试约为1000-2000元。执行时间方面，1000个单元测试通常在10秒内完成，100个集成测试需要1-5分钟，10个E2E测试则可能超过10分钟。反馈速度直接影响到开发效率：单元测试可以在开发者每次保存文件后运行（秒级反馈），集成测试在提交PR时运行（分钟级反馈），E2E测试通常在合并前运行（十分钟级反馈）。

ROI优化建议：如果发现E2E测试的维护成本过高（每次UI变更都需要修改大量测试）、运行时间过长（超过30分钟）、不稳定率过高（超过10%），应当考虑将这些E2E测试下移到集成测试层。反之，如果某个业务场景在单元测试和集成测试中反复无法发现缺陷，而上线后频繁出现问题，则应当考虑为该场景补充E2E测试。ROI的监控是一个持续的过程，建议每个季度进行一次测试审计，评估各层测试的投入产出比。

# 测试分层配置：自动标记测试层级
# conftest.py
import pytest
import os

def pytest_collection_modifyitems(config, items):
    """根据测试文件路径自动添加层级标记"""
    for item in items:
        filepath = item.fspath
        if "/tests/unit/" in str(filepath):
            item.add_marker(pytest.mark.unit)
        elif "/tests/integration/" in str(filepath):
            item.add_marker(pytest.mark.integration)
        elif "/tests/e2e/" in str(filepath):
            item.add_marker(pytest.mark.e2e)

# 按文件目录结构自动分层的好处：
# - 新的测试文件不需要手动添加标记
# - 团队成员可以直观地从文件结构理解测试分层

# 测试ROI分析脚本
import subprocess
import json
from datetime import datetime

class TestROIAnalyzer:
    """测试投资回报率分析"""

    def calculate_test_metrics(self, test_dir: str):
        """计算测试资产指标"""
        result = subprocess.run(
            ["pytest", test_dir, "--collect-only", "-q"],
            capture_output=True, text=True
        )
        total_tests = len(result.stdout.strip().split("\n"))

        # 运行时间测量
        start = datetime.now()
        subprocess.run(["pytest", test_dir, "-q", "--tb=no"],
                       capture_output=True)
        duration = (datetime.now() - start).total_seconds()

        # 按标记分组统计
        markers_result = subprocess.run(
            ["pytest", test_dir, "--markers", "-q"],
            capture_output=True, text=True
        )

        return {
            "total_tests": total_tests,
            "duration_seconds": duration,
            "tests_per_second": total_tests / duration if duration > 0 else 0
        }

    def analyze_roi(self):
        """分析各层测试ROI"""
        layers = {
            "unit": "tests/unit",
            "integration": "tests/integration",
            "e2e": "tests/e2e"
        }
        results = {}
        for layer, path in layers.items():
            if os.path.exists(path):
                results[layer] = self.calculate_test_metrics(path)

        # 输出ROI报告
        print("=" * 60)
        print("测试ROI分析报告")
        print("=" * 60)
        for layer, metrics in results.items():
            print(f"\n{layer.upper()} 测试:")
            print(f"  数量: {metrics['total_tests']}")
            print(f"  耗时: {metrics['duration_seconds']:.2f}s")
            print(f"  速度: {metrics['tests_per_second']:.0f} tests/s")

# 可视化测试分布饼图
import matplotlib.pyplot as plt

def plot_test_distribution(stats: dict):
    """绘制测试分布饼图（用于团队周报）"""
    labels = ["单元测试", "集成测试", "E2E测试"]
    sizes = [stats["unit"], stats["integration"], stats["e2e"]]
    colors = ["#2ecc71", "#3498db", "#e74c3c"]
    explode = (0.05, 0.05, 0.1)  # 突出E2E

    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))

    # 饼图
    ax1.pie(sizes, explode=explode, labels=labels, colors=colors,
            autopct="%1.1f%%", shadow=True, startangle=90)
    ax1.axis("equal")
    ax1.set_title("测试分布比例")

    # 条形图（显示执行时间）
    times = [stats.get(f"{k}_time", 0) for k in ["unit", "integration", "e2e"]]
    ax2.bar(labels, times, color=colors)
    ax2.set_ylabel("执行时间（秒）")
    ax2.set_title("各层测试执行时间")
    for i, v in enumerate(times):
        ax2.text(i, v + 0.1, f"{v:.1f}s", ha="center")

    plt.tight_layout()
    plt.savefig("test_distribution_chart.png", dpi=150)

六、测试数据管理

测试数据管理是测试策略中容易被低估的环节，但它直接影响到测试的可靠性、可重复性和维护成本。一个成熟的测试数据策略需要解决三个核心问题：测试数据如何生成、测试数据如何隔离、测试数据如何清理。在Python项目中，常用的测试数据管理方案包括固定数据（Fixture Data）、随机数据（Random Data）和工厂数据（Factory Data），三种方案各有适用场景。

固定数据策略

固定数据（也称Fixture Data或Seed Data）使用预先定义好的静态数据集，通常以JSON、YAML或SQL文件形式存储在代码仓库中。优点是数据确定性强，测试结果完全可预测；缺点是不够灵活，数据变更需要修改多个测试。固定数据的适用场景包括：测试特定的边界条件（如含特殊字符的用户名）、测试业务规则（如满减活动的金额边界）、回归测试（确保已知场景持续正确）。固定数据的维护需要注意：确保每次Git提交时数据文件和测试代码同步更新，避免"数据漂移"。实践中，建议将固定数据集中管理在tests/fixtures/目录下，按数据类型分文件组织。

随机数据策略

随机数据策略通过数据生成器（如Faker库）创建测试数据。优点是覆盖范围广，容易发现边界值和特殊字符相关问题；缺点是数据不确定性可能导致偶发性测试失败。随机数据的适用场景包括：压力测试和负载测试（需要大量不同数据）、格式验证测试（验证输入验证逻辑是否正确处理各种格式的数据）、探索性测试（补充测试人员手动构造的数据）。使用随机数据的两条铁律：一定要记录随机种子（seed），确保可复现失败；不要在断言中依赖随机生成的具体值，应使用属性断言（如"结果的长度大于0"而非"结果等于42"）。

工厂数据策略

工厂模式（Factory Pattern）是固定数据和随机数据的折中方案，通过工厂类（如factory_boy库）根据模板动态生成测试数据。工厂类可以定义字段的默认值、生成规则、关联关系（如一个用户自动关联一个地址）。优点是灵活性高（每个测试可以定制特定字段）、可维护性好（数据变更只需修改工厂定义）、支持继承（可以通过SubFactory创建关联对象）。工厂模式的实践建议：一个数据库模型对应一个Factory类，工厂类的默认值应产生"有效"数据（即能通过所有验证的数据），测试方法通过覆盖特定字段来构造"无效"数据。

# 固定数据方案：JSON Fixture
# tests/fixtures/users.json
# [
#   {"id": 1, "username": "admin", "role": "admin", "active": true},
#   {"id": 2, "username": "editor", "role": "editor", "active": true},
#   {"id": 3, "username": "viewer", "role": "viewer", "active": false}
# ]

# tests/test_user_filters.py
import json
import pytest

@pytest.fixture
def load_user_fixtures():
    """加载固定测试数据"""
    with open("tests/fixtures/users.json") as f:
        return json.load(f)

def test_filter_active_users(load_user_fixtures):
    """测试过滤活跃用户"""
    from src.services import UserService
    service = UserService()
    active_users = service.filter_active(load_user_fixtures)
    assert len(active_users) == 2
    assert all(u["active"] for u in active_users)

# 随机数据方案：使用Faker
from faker import Faker
import pytest

fake = Faker("zh_CN")
Faker.seed(42)  # 固定随机种子确保可复现

@pytest.fixture
def random_user_data():
    """生成随机用户数据"""
    return {
        "username": fake.user_name(),
        "email": fake.email(),
        "phone": fake.phone_number(),
        "address": fake.address(),
        "birthday": fake.date_of_birth(minimum_age=18, maximum_age=80).isoformat(),
    }

class TestUserRegistration:
    """用户注册测试——使用随机数据"""

    @pytest.mark.parametrize("run", range(20))
    def test_registration_with_random_data(self, run, random_user_data, client):
        """20轮随机数据注册测试"""
        response = client.post("/api/users/register", json=random_user_data)
        assert response.status_code == 201
        # 使用属性断言而非值断言
        assert "id" in response.json()
        assert response.json()["email"] == random_user_data["email"]

    def test_random_phone_number_formats(self, client):
        """验证随机手机号格式——使用属性断言"""
        for _ in range(50):
            phone = fake.phone_number()
            response = client.post("/api/users/validate-phone",
                                   json={"phone": phone})
            # 验证手机号格式是否被正确识别
            assert response.json()["is_valid"] in [True, False]

# 工厂数据方案：使用factory_boy
import factory
from datetime import datetime
from src.models import User, Order, OrderItem

class UserFactory(factory.Factory):
    """用户数据工厂"""
    class Meta:
        model = User

    # 默认生成有效数据
    username = factory.Sequence(lambda n: f"user_{n:04d}")
    email = factory.LazyAttribute(lambda o: f"{o.username}@example.com")
    is_active = True
    role = "user"
    created_at = factory.LazyFunction(datetime.now)

    # 子工厂：关联地址
    @factory.post_generation
    def address(self, create, extracted, **kwargs):
        if not create:
            return
        if extracted:
            self.address = extracted

class OrderFactory(factory.Factory):
    """订单数据工厂"""
    class Meta:
        model = Order

    user = factory.SubFactory(UserFactory)
    total_amount = factory.LazyAttribute(
        lambda o: sum(item.price * item.quantity for item in o.items)
    )
    status = "pending"

class TestOrderWithFactories:
    """使用工厂数据的订单测试"""

    def test_order_with_multiple_items(self, db_session):
        """测试含多个商品条目的订单"""
        user = UserFactory()
        db_session.add(user)
        order = OrderFactory(
            user=user,
            items=[
                OrderItem(name="商品A", price=50.0, quantity=2),
                OrderItem(name="商品B", price=30.0, quantity=1),
            ]
        )
        db_session.add(order)
        db_session.commit()

        # 验证金额自动计算
        assert order.total_amount == 130.0

    def test_order_with_specific_status(self, db_session):
        """测试特定状态的订单"""
        order = OrderFactory(
            status="paid",
            paid_at=factory.LazyFunction(datetime.now)
        )
        db_session.add(order)
        db_session.commit()
        assert order.status == "paid"
        assert order.paid_at is not None

    def test_inactive_user_cannot_create_order(self, db_session):
        """测试非活跃用户不能下单"""
        inactive_user = UserFactory(is_active=False)
        db_session.add(inactive_user)
        db_session.commit()
        from src.services import OrderService
        service = OrderService()
        with pytest.raises(PermissionError):
            service.create_order(inactive_user.id, items=[{"product_id": 1, "qty": 1}])

七、测试运行优化

随着项目规模的增长，测试执行时间会逐渐成为开发效率的瓶颈。一个中等规模的Python项目（500个测试用例）的全量测试可能需要15-30分钟。优化测试运行的核心策略包括：分层运行、并行执行、测试分片（Test Sharding）、构建缓存（Build Cache）和增量测试。目标是让开发者在每次提交代码后，能够在3-5分钟内获得足够的质量反馈，而不是等待30分钟以上的全量测试。

分层运行策略

分层运行是最基础也是最重要的优化策略：在开发者的本地开发阶段，只运行受影响的单元测试（通过pytest-watch或pytest-xdist的自动发现模式）；向远程仓库推送代码（pre-push hook）时，运行单元测试和集成测试；在PR的CI流程中，运行单元测试+集成测试+核心E2E测试；在合并到主分支后，运行全量测试。这种分层策略确保开发者可以在几秒内获得本地修改的反馈（仅运行关联测试），而全面的回归验证在后台的CI系统中异步完成。

并行执行与测试分片

pytest-xdist是Python中最常用的并行执行工具，它可以将测试用例分发到多个CPU核心或远程机器上执行。在本地开发中，使用-n auto参数自动检测CPU核心数量；在CI环境中，可以结合测试分片（Test Sharding）技术进一步缩短执行时间。测试分片的核心思路是将测试用例分成多个组（Shard），每个CI runner执行其中一个分片。相比简单的并行，分片可以将整体执行时间从"最慢分片的耗时"降低到"总耗时/分片数"。合理分片的关键是：让每个分片的测试数量和执行时间大致相等，避免某些分片成为瓶颈。

增量测试

增量测试（Incremental Testing）只运行与当前代码变更相关的测试，而非运行全量测试。在大型项目中，增量测试可以节省60-80%的测试时间。实现增量测试的常见方法包括：使用pytest-testmon自动检测受影响的测试（基于代码覆盖率分析和Git差异对比）；使用pytest-incremental按模块依赖关系确定测试范围；通过自定义脚本解析Git diff，匹配对应的测试文件（如修改了src/services/user.py，则运行tests/unit/services/test_user.py和tests/integration/test_user_api.py）。增量测试的关键问题是要处理间接影响——修改了一个工具函数可能影响数十个调用者，需要确保这些影响也被覆盖。

# 分层运行配置：Makefile
.PHONY: test test-unit test-integration test-e2e test-ci

# 本地开发快速反馈：仅运行单元测试
test-unit:
	pytest tests/unit -x --timeout=30 -v

# 本地提交前检查：单元+集成
test-pr:
	pytest tests/unit tests/integration -x --timeout=60 -v \
		--cov=src --cov-report=term-missing

# CI中完整运行（含E2E）
test-ci:
	pytest -v --run-e2e --timeout=120 \
		--cov=src --cov-report=xml \
		--junitxml=test-results.xml

# 增量测试：仅运行修改文件对应的测试
test-incremental:
	git diff --name-only HEAD~1 | grep "^src/" | \
		sed 's/^src/tests/' | sed 's/\.py$$/_test.py/' | \
		xargs pytest -v --timeout=30

# 并行执行
test-parallel:
	pytest tests/unit -n auto --dist=loadscope -v

# GitHub Actions 测试分片配置
# .github/workflows/test.yml
name: Test Suite

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      fail-fast: false  # 一个分片失败不取消其他分片
      matrix:
        shard: [1, 2, 3, 4]  # 4个分片并行

    steps:
      - uses: actions/checkout@v3

      - name: 设置Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.11"

      - name: 安装依赖
        run: |
          python -m pip install --upgrade pip
          pip install pytest pytest-xdist pytest-cov

      - name: 运行测试（分片 ${{ matrix.shard }}）
        run: |
          pytest tests/unit tests/integration \
            --splits 4 \
            --group ${{ matrix.shard }} \
            --splitting-algorithm least_duration \
            --cov=src \
            --cov-append \
            --junitxml=test-results-${{ matrix.shard }}.xml

      - name: 上传测试结果
        uses: actions/upload-artifact@v3
        with:
          name: test-results-${{ matrix.shard }}
          path: test-results-${{ matrix.shard }}.xml

  # 合并覆盖率报告
  coverage-report:
    needs: [test]
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: 合并覆盖率
        run: |
          pip install coverage
          coverage combine
          coverage report
          coverage html

# 增量测试：pytest-testmon配置
# testmon可以自动识别受代码变更影响的测试
# 安装：pip install pytest-testmon

# 首次运行（建立基准覆盖数据）
# pytest --testmon

# 后续运行（只执行受影响的测试）
# pytest --testmon

# pytest-testmon的工作原理：
# 1. 首次运行时记录每个测试用例覆盖的代码行
# 2. 后续运行时对比Git diff，只执行覆盖了变更行的测试
# 3. 如果新增了未被任何测试覆盖的代码，testmon会报警告

# CI中使用pytest-testmon（需要持久化.testmon数据库）
# .github/workflows/testmon.yml
# steps:
#   - uses: actions/cache@v3
#     with:
#       path: .testmon
#       key: testmon-${{ github.sha }}
#       restore-keys: testmon-
#   - run: pytest --testmon --testmon-forceselect

# 进阶：手动实现增量测试选择器
import subprocess
import re
from pathlib import Path

def get_affected_tests(base_branch="origin/main"):
    """获取受当前变更影响的测试文件"""
    # 获取变更文件列表
    result = subprocess.run(
        ["git", "diff", "--name-only", base_branch],
        capture_output=True, text=True
    )
    changed_files = result.stdout.strip().split("\n")

    # 匹配对应的测试文件
    affected_tests = set()
    for file in changed_files:
        if file.startswith("src/"):
            # 将 src/module/foo.py 映射到 tests/module/test_foo.py
            test_file = re.sub(r"^src/(.*)\.py$", r"tests/\1_test.py", file)
            if Path(test_file).exists():
                affected_tests.add(test_file)
            # 也检查 mock 测试文件
            test_file_alt = re.sub(r"^src/(.*)\.py$", r"tests/test_\1.py", file)
            if Path(test_file_alt).exists():
                affected_tests.add(test_file_alt)

    return list(affected_tests)

八、CI/CD测试阶段

CI/CD流水线中的测试阶段划分直接影响到团队的质量反馈速度和发布效率。一个精心设计的CI/CD测试流水线将测试活动分布在代码管线的不同关卡，从本地预提交检查到生产环境部署前的最终验证，每一关都有明确的测试目标和通过标准。在Python项目中，典型的CI/CD测试流水线包括四个主要阶段：提交前检查（Pre-commit）、PR检查（Pull Request）、合并后验证（Post-merge）和部署前测试（Pre-deploy）。

提交前检查（Pre-commit）

提交前检查的目标是在代码进入版本控制系统之前拦截绝大多数低级问题。这个阶段应该在开发者本地环境运行，确保提交的代码不会破坏基本的质量门禁。典型的pre-commit检查内容包括：代码格式化检查（Black、isort）、Lint检查（Flake8、pylint、mypy类型检查）、运行直接受影响的单元测试（增量测试）。Pre-commit阶段的目标是"快"和"准"——整个流程应在30秒内完成，只检查与本次提交直接相关的代码。建议使用pre-commit工具管理Hook脚本，而不是手动配置.git/hooks/目录。

PR检查（GitHub Actions）

PR（Pull Request）检查是CI/CD流水线的核心环节，也是质量保障的最重要关卡。当开发者创建或更新PR时，CI系统自动触发一整套检查流程。典型的PR检查流程包括：全量单元测试+集成测试、代码覆盖率检查（增量覆盖率门禁80%+）、代码质量检测（SonarQube或CodeClimate）、安全检查（Bandit或Safety）。PR检查的设计要点：总执行时间控制在10-15分钟以内（使用测试分片和缓存优化）；分级门禁策略（单元测试必须全部通过，覆盖率不达标发出警告而非阻断，安全检查必须全部通过）；提交CI状态回PR，让Reviewer直观了解代码质量状况。

部署前测试

部署前测试是代码上线前的最后一道质量防线，运行在预发布环境（Staging）或金丝雀发布环境（Canary）中。这一阶段的测试包括：E2E测试（在预发布环境上运行完整的端到端测试）、性能测试（验证关键API的响应时间在可接受范围内）、兼容性测试（验证不同浏览器、设备或Python版本的兼容性）、数据验证（执行数据库迁移脚本并验证数据完整性）。部署前测试通常作为发布流程的手动触发步骤，由发布经理确认预发布环境的测试全部通过后进行正式发布。在实践中，建议部署前测试的E2E数量控制在20个以内，运行时间不超过30分钟。

# Pre-commit配置：.pre-commit-config.yaml
repos:
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.4.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-yaml
      - id: check-json
      - id: check-added-large-files
        args: ["--maxkb=500"]

  - repo: https://github.com/psf/black
    rev: 23.1.0
    hooks:
      - id: black
        args: [--line-length=88]

  - repo: https://github.com/PyCQA/isort
    rev: 5.12.0
    hooks:
      - id: isort
        args: [--profile=black]

  - repo: https://github.com/PyCQA/flake8
    rev: 6.0.0
    hooks:
      - id: flake8
        args: [--max-line-length=88, --extend-ignore=E203]

  - repo: local
    hooks:
      - id: pytest-incremental
        name: 运行增量单元测试
        entry: pytest tests/unit --testmon --testmon-forceselect
        language: system
        pass_filenames: false
        always_run: true
        stages: [commit]

  - repo: https://github.com/PyCQA/bandit
    rev: 1.7.5
    hooks:
      - id: bandit
        args: [-r, src, -x, tests]

# GitHub Actions PR检查完整流水线
name: PR Quality Check

on:
  pull_request:
    branches: [main, develop]

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: "3.11"
          cache: "pip"
      - run: pip install black isort flake8 mypy
      - run: black --check src tests
      - run: isort --check-only --profile black src tests
      - run: flake8 src tests --max-line-length=88
      - run: mypy src --ignore-missing-imports

  unit-integration-tests:
    runs-on: ubuntu-latest
    needs: [lint]
    strategy:
      matrix:
        python-version: ["3.9", "3.10", "3.11"]

    services:
      postgres:
        image: postgres:14-alpine
        env:
          POSTGRES_DB: test_db
          POSTGRES_USER: test_user
          POSTGRES_PASSWORD: test_pass
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: ${{ matrix.python-version }}
          cache: "pip"

      - name: Install dependencies
        run: |
          pip install pytest pytest-xdist pytest-cov
          pip install -r requirements-dev.txt

      - name: Run unit & integration tests
        run: |
          pytest tests/unit tests/integration \
            -n auto --dist=loadscope \
            --cov=src --cov-report=xml \
            --junitxml=test-results.xml
        env:
          DATABASE_URL: postgresql://test_user:test_pass@localhost:5432/test_db

      - name: Upload coverage
        uses: codecov/codecov-action@v3

  security-scan:
    runs-on: ubuntu-latest
    needs: [lint]
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - run: pip install bandit safety
      - run: bandit -r src -f json -o bandit-report.json
      - run: safety check -r requirements.txt

  e2e-tests:
    runs-on: ubuntu-latest
    needs: [unit-integration-tests]
    if: github.event.pull_request.draft == false
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - run: pip install pytest playwright
      - run: playwright install chromium
      - run: pytest tests/e2e --run-e2e --timeout=60000
        env:
          TEST_URL: https://staging.example.com

# 部署前测试检查清单脚本
# scripts/pre_deploy_check.py
import subprocess
import sys
from datetime import datetime

def run_test_suite():
    """执行部署前测试检查"""
    checks = {
        "E2E测试": "pytest tests/e2e --run-e2e -q --tb=short",
        "性能测试": "pytest tests/performance -q --tb=short",
        "数据库迁移测试": "pytest tests/migration -q --tb=short",
        "安全审计": "bandit -r src -q",
    }

    results = {}
    all_passed = True

    print(f"\n{'='*60}")
    print(f"部署前测试检查 - {datetime.now()}")
    print(f"{'='*60}\n")

    for name, command in checks.items():
        print(f"[运行] {name}...", end=" ")
        sys.stdout.flush()

        result = subprocess.run(
            command, shell=True, capture_output=True, text=True
        )

        if result.returncode == 0:
            print("\033[92m通过\033[0m")
            results[name] = True
        else:
            print("\033[91m失败\033[0m")
            print(f"  输出: {result.stdout[-500:]}")
            print(f"  错误: {result.stderr[-500:]}")
            results[name] = False
            all_passed = False

    print(f"\n{'='*60}")
    print(f"结果: {'全部通过' if all_passed else '存在失败'}")
    for name, passed in results.items():
        status = "\033[92m通过\033[0m" if passed else "\033[91m失败\033[0m"
        print(f"  {name}: {status}")

    return all_passed

if __name__ == "__main__":
    success = run_test_suite()
    sys.exit(0 if success else 1)

九、实战案例

理论知识的最终价值在于指导实践。以下通过三个不同项目类型的实战案例，展示如何将前文所述的测试策略原则应用到实际项目中。每个案例都包含项目背景、测试策略设计、具体实施步骤和关键决策点。

案例一：Web应用测试策略

背景：一个基于FastAPI的电商后端服务，包含用户认证、商品管理、购物车、订单处理、支付集成等功能模块。团队6人，两周一个迭代。测试策略设计要点：单元测试覆盖核心业务逻辑（价格计算、库存扣减、优惠券验证），使用pytest+factory_boy；集成测试覆盖所有API端点（使用FastAPI TestClient），验证请求路由、参数校验、响应格式、数据库交互；E2E测试覆盖4条关键路径（用户注册→登录→搜索→下单→支付→查看订单）。技术选型：pytest + pytest-cov + pytest-xdist + factory_boy + pytest-httpx + Playwright。分层比例：单元测试72%，集成测试23%，E2E测试5%。覆盖率目标：核心业务90%，整体80%。

# Web应用测试策略实战：分层测试结构
# tests/
# ├── conftest.py               # 共享fixture和配置
# ├── unit/
# │   ├── conftest.py           # 单元测试专属fixture（Mock）
# │   ├── services/
# │   │   ├── test_cart.py      # 购物车业务逻辑
# │   │   ├── test_order.py     # 订单处理逻辑
# │   │   ├── test_discount.py  # 折扣计算逻辑
# │   │   └── test_inventory.py # 库存管理逻辑
# │   └── models/
# │       ├── test_user.py      # 用户模型方法
# │       └── test_product.py   # 商品模型方法
# ├── integration/
# │   ├── conftest.py           # 集成测试fixture（数据库、HTTP客户端）
# │   ├── api/
# │   │   ├── test_user_api.py       # 用户API
# │   │   ├── test_product_api.py    # 商品API
# │   │   ├── test_cart_api.py       # 购物车API
# │   │   ├── test_order_api.py      # 订单API
# │   │   └── test_payment_api.py    # 支付API
# │   └── db/
# │       ├── test_migrations.py     # 数据库迁移测试
# │       └── test_queries.py        # 复杂查询测试
# └── e2e/
#     ├── conftest.py           # E2E测试fixture（浏览器、测试用户）
#     ├── pages/                # Page Objects
#     │   ├── login_page.py
#     │   ├── search_page.py
#     │   ├── cart_page.py
#     │   └── order_page.py
#     └── test_flows.py         # 4条关键路径E2E测试

# 核心API集成测试示例
def test_create_order_with_inventory_check(client, db_session, user_factory, product_factory):
    """创建订单时自动扣减库存的集成测试"""
    # 准备数据
    user = user_factory()
    product = product_factory(stock=10, price=99.99)
    db_session.add_all([user, product])
    db_session.commit()

    # 下单
    response = client.post(
        "/api/orders",
        json={
            "user_id": user.id,
            "items": [{"product_id": product.id, "quantity": 3}]
        },
        headers={"Authorization": f"Bearer {user.token}"}
    )

    # 验证订单创建成功
    assert response.status_code == 201
    order_data = response.json()
    assert order_data["total_amount"] == 299.97  # 99.99 * 3

    # 验证库存扣减
    db_session.refresh(product)
    assert product.stock == 7  # 10 - 3

    # 验证库存不足时拒绝下单
    response = client.post(
        "/api/orders",
        json={
            "user_id": user.id,
            "items": [{"product_id": product.id, "quantity": 10}]
        },
        headers={"Authorization": f"Bearer {user.token}"}
    )
    assert response.status_code == 400
    assert "库存不足" in response.json()["detail"]

案例二：微服务测试策略

背景：基于Python的微服务架构，包含6个独立服务（用户服务、商品服务、订单服务、支付服务、通知服务、网关服务），服务间通过gRPC和消息队列（RabbitMQ）通信。团队12人，三个Scrum团队。测试策略设计要点：每个服务独立维护自己的测试套件，单元测试覆盖各服务的内部逻辑；集成测试关注服务与基础设施（数据库、消息队列、缓存）的交互；契约测试是微服务测试的核心——每个服务作为消费者编写契约，作为提供者验证契约，确保服务间API兼容；E2E测试只覆盖跨服务的核心业务场景（如"用户下单"流程涉及4个服务的协作）。关键决策：引入Pact Broker管理契约版本，使用Testcontainers管理消息队列的测试环境，每个服务在CI中独立验证后再进行跨服务集成验证。

# 微服务契约测试示例（使用pact-python）
import pytest
from pact import Consumer, Provider, Like, Term

# 定义消费者契约
pact = Consumer("订单服务").has_pact_with(
    Provider("支付服务"),
    pact_dir="./pacts"
)

@pytest.fixture
def payment_pact():
    """创建支付服务消费者契约"""
    pact.start_service()
    yield pact
    pact.stop_service()

class TestOrderPaymentContract:
    """订单服务作为消费者定义的支付服务契约"""

    def test_successful_payment(self, payment_pact):
        """支付成功场景契约"""
        # 定义消费者期望的交互
        expected = {
            "status": "success",
            "transaction_id": "TXN-20250301-0001",
            "amount": 299.99,
            "currency": "CNY",
            "paid_at": Like("2025-03-01T12:00:00Z")
        }

        (payment_pact
         .given("支付渠道正常")
         .upon_receiving("处理支付请求")
         .with_request(
             method="POST",
             path="/api/v1/payments/charge",
             headers={"Content-Type": "application/json"},
             body={
                 "order_id": "ORDER-001",
                 "amount": 299.99,
                 "currency": "CNY",
                 "payment_method": "alipay"
             }
         )
         .will_respond_with(200, headers={
             "Content-Type": "application/json"
         }, body=expected))

        # 执行契约验证
        with payment_pact:
            result = payment_pact.call_service(
                method="POST",
                path="/api/v1/payments/charge",
                headers={"Content-Type": "application/json"},
                body={
                    "order_id": "ORDER-001",
                    "amount": 299.99,
                    "currency": "CNY",
                    "payment_method": "alipay"
                }
            )
            assert result["status"] == "success"
            assert result["transaction_id"] is not None

    def test_payment_insufficient_balance(self, payment_pact):
        """余额不足场景契约"""
        (payment_pact
         .given("用户余额不足")
         .upon_receiving("处理支付请求（余额不足）")
         .with_request(
             method="POST",
             path="/api/v1/payments/charge",
             headers={"Content-Type": "application/json"},
             body={
                 "order_id": "ORDER-002",
                 "amount": 9999.99,
                 "currency": "CNY",
                 "payment_method": "balance"
             }
         )
         .will_respond_with(402, headers={
             "Content-Type": "application/json"
         }, body={
             "status": "failed",
             "error": "余额不足",
             "code": "INSUFFICIENT_BALANCE"
         }))

案例三：数据处理项目测试策略

背景：基于Python的数据ETL和处理管线项目，从多个数据源（MySQL、MongoDB、第三方API）抽取数据，进行清洗、转换、聚合后写入数据仓库（ClickHouse）。每天处理约1亿条记录。测试策略设计要点：单元测试覆盖数据转换函数、验证逻辑和聚合算法，使用pytest参数化测试大量边界条件；组件测试验证单个处理阶段（Extract、Transform、Load），使用小型真实数据集运行完整的数据处理流程；端到端测试（这里更准确地说是"集成管线测试"）使用生产脱敏数据运行完整的ETL管线，验证数据的一致性（输入数据量和输出数据量对比、关键指标聚合值对比）。关键决策：建立"黄金数据集"——一个包含各种边界情况和异常数据的预定义数据集，用于回归测试数据处理逻辑；数据一致性校验使用双运行对比策略（将新版本的输出与旧版本的基线输出进行对比）。

# 数据处理项目ETL测试示例
import pytest
import pandas as pd
from datetime import datetime
from src.transforms import DataTransformer

class TestDataTransformer:
    """数据转换函数的单元测试"""

    @pytest.mark.parametrize("input_val,expected", [
        ("  hello  ", "hello"),
        ("您好 世界", "您好 世界"),
        ("  ", ""),
        (None, ""),
        ("ABC", "abc"),
        ("测试DATA", "测试data"),
    ])
    def test_clean_string(self, input_val, expected):
        """测试字符串清洗函数"""
        transformer = DataTransformer()
        result = transformer.clean_string(input_val)
        assert result == expected

    def test_transform_sales_data(self, spark):
        """测试销售数据转换逻辑"""
        transformer = DataTransformer()

        input_data = [
            {"date": "2025-01-01", "product": "A", "amount": "100.50", "qty": "2"},
            {"date": "2025-01-01", "product": "B", "amount": "50.25", "qty": "5"},
            {"date": "invalid", "product": "C", "amount": "abc", "qty": "-1"},
        ]

        result = transformer.transform_sales(input_data)

        # 有效数据行
        assert len(result["valid_rows"]) == 2
        assert result["valid_rows"][0]["product_normalized"] == "a"

        # 异常数据行
        assert len(result["error_rows"]) == 1
        assert result["error_rows"][0]["reason"] is not None

    def test_aggregation_correctness(self):
        """测试聚合计算的正确性——与基线对比"""
        transformer = DataTransformer()

        input_df = pd.DataFrame({
            "category": ["A", "A", "B", "B", "C"],
            "value": [10, 20, 30, 40, 50],
            "weight": [1, 2, 1, 2, 1],
        })

        result = transformer.weighted_average(input_df, "value", "weight")

        # 手动计算基线：(10*1 + 20*2 + 30*1 + 40*2 + 50*1) / (1+2+1+2+1)
        expected = (10 + 40 + 30 + 80 + 50) / 7
        assert abs(result - expected) < 0.0001

# 数据处理管线端到端测试
import pytest
from pathlib import Path

class TestETLPipeline:
    """ETL管线端到端测试——使用黄金数据集"""

    GOLDEN_DATASET = Path("tests/fixtures/golden_dataset/")

    @pytest.fixture
    def golden_input_data(self):
        """加载黄金输入数据集"""
        return pd.read_parquet(
            self.GOLDEN_DATASET / "input_data.parquet"
        )

    @pytest.fixture
    def golden_expected_output(self):
        """加载基线输出数据"""
        return pd.read_parquet(
            self.GOLDEN_DATASET / "expected_output.parquet"
        )

    def test_full_etl_pipeline(self, golden_input_data, golden_expected_output):
        """完整ETL管线测试"""

        # 1. Extract阶段
        extractor = DataExtractor()
        raw_data = extractor.extract_from_memory(golden_input_data)
        assert len(raw_data) > 0

        # 2. Transform阶段
        transformer = DataTransformer()
        transformed_data = transformer.transform_sales(raw_data)

        # 验证转换后的数据量
        assert len(transformed_data["valid_rows"]) > 0
        assert len(transformed_data["valid_rows"]) == len(golden_expected_output)

        # 3. Load阶段
        loader = DataLoader(target="memory")
        result = loader.load(transformed_data["valid_rows"])

        # 4. 数据一致性校验
        loaded_df = result["dataframe"]

        # 关键指标对比
        assert loaded_df["total_amount"].sum() == pytest.approx(
            golden_expected_output["total_amount"].sum(), rel=0.01
        )

        # 行数一致
        assert len(loaded_df) == len(golden_expected_output)

        # 唯一键一致
        assert set(loaded_df["order_id"]) == set(
            golden_expected_output["order_id"]
        )

        print(f"ETL管线验证通过: 输入{len(raw_data)}行 -> "
              f"有效{len(transformed_data['valid_rows'])}行 -> "
              f"写入{len(loaded_df)}行")