环境声明

• Python 版本：Python 3.12+
• 开发工具：PyCharm 或 VS Code
• 操作系统：Windows / macOS / Linux (通用)
• 相关库：
  • pytest (8.0+)
  • pytest-cov (5.0+)
  • pytest-asyncio (0.23+)
  • httpx (0.27+) - 用于异步测试
  • factory-boy (3.3+) - 测试数据工厂

---

学习目标

1. 掌握 pytest 测试框架的核心用法
2. 学会编写单元测试和集成测试
3. 掌握测试客户端的使用和模拟对象
4. 学会数据库测试和事务隔离
5. 理解测试覆盖率的概念和优化方法

---

1. 测试基础概念

1.1 为什么要写测试

想象你是一名建筑工程师：

• 不写测试：盖完楼直接让人住，出了问题再修（代价巨大）
• 写测试：每盖一层都检查，确保稳固再继续（代价小，信心足）

一句话总结：测试是代码的"安全带"，让你在重构时心里有底。

1.2 测试金字塔

        /\
       /  \      E2E 测试（端到端）
      /----\     数量少，成本高
     /      \
    /--------\   集成测试
   /          \  数量中等
  /------------\
 /              \ 单元测试
/----------------\ 数量多，成本低

测试类型	范围	速度	成本	稳定性
单元测试	单个函数/类	快	低	高
集成测试	多个组件协作	中等	中等	中等
E2E 测试	完整用户流程	慢	高	低

---

2. pytest 基础

2.1 安装与配置

# 安装 pytest 及相关插件
pip install pytest pytest-cov pytest-asyncio pytest-mock factory-boy

# 验证安装
pytest --version

2.2 第一个测试

# test_calculator.py

def add(a, b):
    """加法函数"""
    return a + b

def subtract(a, b):
    """减法函数"""
    return a - b

# 测试函数：以 test_ 开头
class TestCalculator:
    """计算器测试类"""
    
    def test_add_positive_numbers(self):
        """测试正数相加"""
        assert add(2, 3) == 5
    
    def test_add_negative_numbers(self):
        """测试负数相加"""
        assert add(-2, -3) == -5
    
    def test_add_mixed_numbers(self):
        """测试正负混合相加"""
        assert add(-2, 3) == 1
    
    def test_subtract(self):
        """测试减法"""
        assert subtract(5, 3) == 2
        assert subtract(3, 5) == -2

运行测试：

# 运行所有测试
pytest

# 运行特定文件
pytest test_calculator.py

# 运行特定类
pytest test_calculator.py::TestCalculator

# 运行特定测试方法
pytest test_calculator.py::TestCalculator::test_add_positive_numbers

# 显示详细信息
pytest -v

# 显示打印输出
pytest -s

2.3 常用断言

import pytest

class TestAssertions:
    """断言示例"""
    
    def test_equal(self):
        """相等断言"""
        assert 1 + 1 == 2
    
    def test_not_equal(self):
        """不等断言"""
        assert 1 + 1 != 3
    
    def test_true_false(self):
        """布尔断言"""
        assert True
        assert not False
    
    def test_in(self):
        """包含断言"""
        assert "hello" in "hello world"
        assert 1 in [1, 2, 3]
    
    def test_is_none(self):
        """空值断言"""
        value = None
        assert value is None
    
    def test_raises(self):
        """异常断言"""
        with pytest.raises(ZeroDivisionError):
            1 / 0
    
    def test_raises_with_message(self):
        """异常消息断言"""
        with pytest.raises(ValueError, match="invalid"):
            raise ValueError("invalid input")
    
    def test_almost_equal(self):
        """浮点数比较"""
        assert 0.1 + 0.2 == pytest.approx(0.3)

2.4 Fixture：测试的"脚手架"

import pytest

# 基础 fixture
@pytest.fixture
def sample_data():
    """提供测试数据"""
    return {"name": "张三", "age": 25, "city": "北京"}

# 作用域 fixture
@pytest.fixture(scope="module")
def database_connection():
    """模块级 fixture，只执行一次"""
    print("\n建立数据库连接")
    conn = {"connected": True, "id": 12345}
    yield conn  # 提供资源
    print("\n关闭数据库连接")
    conn["connected"] = False

# 自动使用的 fixture
@pytest.fixture(autouse=True)
def setup_teardown():
    """每个测试前后自动执行"""
    print("\n测试前准备")
    yield
    print("\n测试后清理")

class TestFixtures:
    """Fixture 使用示例"""
    
    def test_use_sample_data(self, sample_data):
        """使用 sample_data fixture"""
        assert sample_data["name"] == "张三"
        assert sample_data["age"] == 25
    
    def test_database(self, database_connection):
        """使用数据库连接"""
        assert database_connection["connected"] is True
    
    def test_multiple_fixtures(self, sample_data, database_connection):
        """使用多个 fixture"""
        assert sample_data is not None
        assert database_connection["id"] == 12345

2.5 参数化测试

import pytest

def is_even(n):
    return n % 2 == 0

class TestParametrized:
    """参数化测试示例"""
    
    # 基础参数化
    @pytest.mark.parametrize("input,expected", [
        (2, True),
        (4, True),
        (3, False),
        (5, False),
        (0, True),
        (-2, True),
    ])
    def test_is_even(self, input, expected):
        """测试偶数判断"""
        assert is_even(input) == expected
    
    # 多参数参数化
    @pytest.mark.parametrize("a,b,expected", [
        (1, 2, 3),
        (5, 5, 10),
        (-1, 1, 0),
    ])
    def test_add(self, a, b, expected):
        """测试加法"""
        assert a + b == expected
    
    # 组合参数化
    @pytest.mark.parametrize("x", [1, 2])
    @pytest.mark.parametrize("y", ["a", "b"])
    def test_combinations(self, x, y):
        """测试所有组合：1a, 1b, 2a, 2b"""
        print(f"\n测试组合: x={x}, y={y}")
        assert isinstance(x, int)
        assert isinstance(y, str)

---

3. Web应用测试

3.1 Flask 应用测试

# app.py - 被测试的应用
from flask import Flask, jsonify, request

app = Flask(__name__)

users = {}

@app.route('/')
def index():
    return jsonify({"message": "Hello, World!"})

@app.route('/users', methods=['GET'])
def get_users():
    return jsonify(list(users.values()))

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = users.get(user_id)
    if user:
        return jsonify(user)
    return jsonify({"error": "User not found"}), 404

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    user_id = len(users) + 1
    user = {
        "id": user_id,
        "name": data.get("name"),
        "email": data.get("email")
    }
    users[user_id] = user
    return jsonify(user), 201

@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    if user_id in users:
        del users[user_id]
        return jsonify({"message": "User deleted"})
    return jsonify({"error": "User not found"}), 404

# test_app.py - 测试文件
import pytest
from app import app

@pytest.fixture
def client():
    """创建测试客户端"""
    app.config['TESTING'] = True
    with app.test_client() as client:
        yield client

@pytest.fixture(autouse=True)
def reset_users():
    """每个测试前重置用户数据"""
    from app import users
    users.clear()
    yield
    users.clear()

class TestFlaskApp:
    """Flask 应用测试"""
    
    def test_index(self, client):
        """测试首页"""
        response = client.get('/')
        assert response.status_code == 200
        assert response.json == {"message": "Hello, World!"}
    
    def test_get_empty_users(self, client):
        """测试获取空用户列表"""
        response = client.get('/users')
        assert response.status_code == 200
        assert response.json == []
    
    def test_create_user(self, client):
        """测试创建用户"""
        response = client.post('/users', json={
            "name": "张三",
            "email": "zhangsan@example.com"
        })
        assert response.status_code == 201
        assert response.json["name"] == "张三"
        assert response.json["email"] == "zhangsan@example.com"
        assert "id" in response.json
    
    def test_get_user(self, client):
        """测试获取单个用户"""
        # 先创建用户
        client.post('/users', json={"name": "李四", "email": "lisi@example.com"})
        
        # 获取用户
        response = client.get('/users/1')
        assert response.status_code == 200
        assert response.json["name"] == "李四"
    
    def test_get_nonexistent_user(self, client):
        """测试获取不存在的用户"""
        response = client.get('/users/999')
        assert response.status_code == 404
        assert "error" in response.json
    
    def test_delete_user(self, client):
        """测试删除用户"""
        # 创建并删除用户
        client.post('/users', json={"name": "王五", "email": "wangwu@example.com"})
        response = client.delete('/users/1')
        assert response.status_code == 200
        
        # 确认已删除
        response = client.get('/users/1')
        assert response.status_code == 404

3.2 FastAPI 应用测试

# main.py - FastAPI 应用
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: str
    age: Optional[int] = None

class User(UserCreate):
    id: int

# 内存存储
users_db = {}
next_id = 1

@app.get("/")
async def root():
    return {"message": "Hello FastAPI"}

@app.get("/users", response_model=List[User])
async def get_users():
    return list(users_db.values())

@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    return users_db[user_id]

@app.post("/users", response_model=User, status_code=201)
async def create_user(user: UserCreate):
    global next_id
    new_user = User(id=next_id, **user.dict())
    users_db[next_id] = new_user
    next_id += 1
    return new_user

@app.put("/users/{user_id}", response_model=User)
async def update_user(user_id: int, user_update: UserCreate):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    
    updated_user = User(id=user_id, **user_update.dict())
    users_db[user_id] = updated_user
    return updated_user

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    del users_db[user_id]
    return {"message": "User deleted"}

# test_main.py - 测试文件
import pytest
from fastapi.testclient import TestClient
from main import app, users_db, next_id

@pytest.fixture
def client():
    """创建测试客户端"""
    return TestClient(app)

@pytest.fixture(autouse=True)
def reset_db():
    """重置数据库"""
    global next_id
    users_db.clear()
    next_id = 1
    yield
    users_db.clear()
    next_id = 1

class TestFastAPIApp:
    """FastAPI 应用测试"""
    
    def test_root(self, client):
        """测试根路径"""
        response = client.get("/")
        assert response.status_code == 200
        assert response.json() == {"message": "Hello FastAPI"}
    
    def test_create_user(self, client):
        """测试创建用户"""
        response = client.post("/users", json={
            "name": "张三",
            "email": "zhangsan@example.com",
            "age": 25
        })
        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "张三"
        assert data["email"] == "zhangsan@example.com"
        assert data["age"] == 25
        assert data["id"] == 1
    
    def test_create_user_validation(self, client):
        """测试创建用户参数验证"""
        # 缺少必填字段
        response = client.post("/users", json={"name": "张三"})
        assert response.status_code == 422
        
        # 邮箱格式错误
        response = client.post("/users", json={
            "name": "张三",
            "email": "invalid-email"
        })
        assert response.status_code == 422
    
    def test_get_users(self, client):
        """测试获取用户列表"""
        # 创建两个用户
        client.post("/users", json={"name": "用户1", "email": "u1@test.com"})
        client.post("/users", json={"name": "用户2", "email": "u2@test.com"})
        
        response = client.get("/users")
        assert response.status_code == 200
        assert len(response.json()) == 2
    
    def test_update_user(self, client):
        """测试更新用户"""
        # 创建用户
        client.post("/users", json={"name": "旧名字", "email": "old@test.com"})
        
        # 更新用户
        response = client.put("/users/1", json={
            "name": "新名字",
            "email": "new@test.com"
        })
        assert response.status_code == 200
        assert response.json()["name"] == "新名字"
        assert response.json()["email"] == "new@test.com"
    
    def test_delete_user(self, client):
        """测试删除用户"""
        client.post("/users", json={"name": "待删除", "email": "del@test.com"})
        
        response = client.delete("/users/1")
        assert response.status_code == 200
        
        # 确认已删除
        response = client.get("/users/1")
        assert response.status_code == 404

---

4. 模拟与依赖注入

4.1 使用 unittest.mock

# services.py
import requests
import time

class PaymentService:
    """支付服务"""
    
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.payment.com/v1"
    
    def charge(self, amount, card_number):
        """处理支付"""
        response = requests.post(
            f"{self.base_url}/charge",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"amount": amount, "card": card_number}
        )
        response.raise_for_status()
        return response.json()

class EmailService:
    """邮件服务"""
    
    def send(self, to, subject, body):
        """发送邮件（模拟耗时操作）"""
        time.sleep(2)  # 模拟网络延迟
        print(f"发送邮件到 {to}: {subject}")
        return True

# test_services.py
import pytest
from unittest.mock import Mock, patch, MagicMock
from services import PaymentService, EmailService

class TestPaymentService:
    """支付服务测试"""
    
    @patch('services.requests')
    def test_charge_success(self, mock_requests):
        """测试支付成功"""
        # 设置模拟返回值
        mock_response = Mock()
        mock_response.json.return_value = {
            "status": "success",
            "transaction_id": "txn_12345"
        }
        mock_response.raise_for_status = Mock()
        mock_requests.post.return_value = mock_response
        
        # 执行测试
        service = PaymentService("test_api_key")
        result = service.charge(100.00, "4111111111111111")
        
        # 验证
        assert result["status"] == "success"
        assert result["transaction_id"] == "txn_12345"
        
        # 验证调用参数
        mock_requests.post.assert_called_once()
        call_args = mock_requests.post.call_args
        assert call_args[0][0] == "https://api.payment.com/v1/charge"
    
    @patch('services.requests')
    def test_charge_failure(self, mock_requests):
        """测试支付失败"""
        mock_requests.post.side_effect = Exception("Network error")
        
        service = PaymentService("test_api_key")
        
        with pytest.raises(Exception, match="Network error"):
            service.charge(100.00, "4111111111111111")

class TestEmailService:
    """邮件服务测试"""
    
    @patch('services.time.sleep')
    def test_send_email(self, mock_sleep):
        """测试发送邮件（不实际等待）"""
        service = EmailService()
        result = service.send("user@example.com", "测试邮件", "内容")
        
        assert result is True
        mock_sleep.assert_called_once_with(2)  # 验证 sleep 被调用

4.2 pytest-mock 插件

import pytest

class TestWithPytestMock:
    """使用 pytest-mock 的测试"""
    
    def test_mock_object(self, mocker):
        """使用 mocker fixture"""
        # 创建 mock 对象
        mock_func = mocker.Mock()
        mock_func.return_value = 42
        
        result = mock_func("arg1", "arg2")
        
        assert result == 42
        mock_func.assert_called_once_with("arg1", "arg2")
    
    def test_patch_with_mocker(self, mocker):
        """使用 mocker.patch"""
        # 模拟内置函数
        mock_open = mocker.patch('builtins.open')
        mock_file = mocker.Mock()
        mock_file.read.return_value = 'file content'
        mock_open.return_value.__enter__.return_value = mock_file
        
        with open('test.txt', 'r') as f:
            content = f.read()
        
        assert content == 'file content'
    
    def test_spy(self, mocker):
        """使用 spy 监视函数"""
        class Calculator:
            def add(self, a, b):
                return a + b
        
        calc = Calculator()
        spy = mocker.spy(calc, 'add')
        
        result = calc.add(2, 3)
        
        assert result == 5
        spy.assert_called_once_with(2, 3)
    
    def test_stub(self, mocker):
        """使用 stub 替换方法"""
        class Database:
            def query(self, sql):
                # 实际会查询数据库
                pass
        
        db = Database()
        mocker.patch.object(db, 'query', return_value=[{"id": 1, "name": "test"}])
        
        result = db.query("SELECT * FROM users")
        assert result == [{"id": 1, "name": "test"}]

4.3 FastAPI 依赖注入测试

# dependencies.py
from fastapi import Header, HTTPException, Depends

async def verify_token(x_token: str = Header(...)):
    """验证 token"""
    if x_token != "secret-token":
        raise HTTPException(status_code=403, detail="Invalid token")
    return x_token

async def get_current_user(token: str = Depends(verify_token)):
    """获取当前用户"""
    # 实际会从数据库查询
    return {"id": 1, "name": "Admin", "token": token}

# app_with_deps.py
from fastapi import FastAPI, Depends
from dependencies import get_current_user

app = FastAPI()

@app.get("/protected")
async def protected_route(user: dict = Depends(get_current_user)):
    return {"message": f"Hello {user['name']}", "user": user}

# test_dependencies.py
import pytest
from fastapi.testclient import TestClient
from app_with_deps import app
from dependencies import get_current_user

@pytest.fixture
def client():
    return TestClient(app)

class TestWithDependencyOverride:
    """测试依赖注入"""
    
    def test_protected_route_without_token(self, client):
        """测试没有 token 时访问受保护路由"""
        response = client.get("/protected")
        assert response.status_code == 403
    
    def test_protected_route_with_token(self, client):
        """测试有 token 时访问受保护路由"""
        response = client.get("/protected", headers={"X-Token": "secret-token"})
        assert response.status_code == 200
        assert response.json()["message"] == "Hello Admin"
    
    def test_override_dependency(self, client):
        """测试覆盖依赖"""
        # 创建模拟用户
        def mock_get_current_user():
            return {"id": 999, "name": "TestUser"}
        
        # 覆盖依赖
        app.dependency_overrides[get_current_user] = mock_get_current_user
        
        try:
            # 现在不需要 token 了
            response = client.get("/protected")
            assert response.status_code == 200
            assert response.json()["message"] == "Hello TestUser"
            assert response.json()["user"]["id"] == 999
        finally:
            # 清理覆盖
            app.dependency_overrides.clear()

---

5. 数据库测试

5.1 使用 SQLite 内存数据库

# database.py
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    email = Column(String, unique=True, nullable=False)

# 创建引擎
engine = create_engine("sqlite:///./test.db")
Base.metadata.create_all(bind=engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# test_database.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from database import Base, User

# 使用内存数据库
TEST_DATABASE_URL = "sqlite:///:memory:"

@pytest.fixture(scope="function")
def db_session():
    """为每个测试创建独立的数据库会话"""
    # 创建内存数据库引擎
    engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
    
    # 创建所有表
    Base.metadata.create_all(bind=engine)
    
    # 创建会话
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    session = TestingSessionLocal()
    
    yield session
    
    # 清理
    session.close()
    Base.metadata.drop_all(bind=engine)

class TestDatabase:
    """数据库测试"""
    
    def test_create_user(self, db_session):
        """测试创建用户"""
        user = User(name="张三", email="zhangsan@example.com")
        db_session.add(user)
        db_session.commit()
        
        # 查询验证
        result = db_session.query(User).filter(User.name == "张三").first()
        assert result is not None
        assert result.email == "zhangsan@example.com"
    
    def test_user_isolation(self, db_session):
        """测试数据库隔离：每个测试有独立数据"""
        # 这个测试开始时数据库是空的
        count = db_session.query(User).count()
        assert count == 0
        
        # 添加数据
        db_session.add(User(name="李四", email="lisi@example.com"))
        db_session.commit()
        
        assert db_session.query(User).count() == 1
    
    def test_unique_constraint(self, db_session):
        """测试唯一约束"""
        # 创建第一个用户
        user1 = User(name="用户1", email="test@example.com")
        db_session.add(user1)
        db_session.commit()
        
        # 尝试创建相同邮箱的用户
        user2 = User(name="用户2", email="test@example.com")
        db_session.add(user2)
        
        with pytest.raises(Exception):
            db_session.commit()

5.2 使用 Factory Boy 生成测试数据

# factories.py
import factory
from factory.alchemy import SQLAlchemyModelFactory
from sqlalchemy.orm import Session
from database import User, Base
from sqlalchemy import create_engine

# 创建内存数据库
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(bind=engine)
SessionLocal = sessionmaker(bind=engine)

class UserFactory(SQLAlchemyModelFactory):
    """用户工厂"""
    class Meta:
        model = User
        sqlalchemy_session = SessionLocal()
        sqlalchemy_session_persistence = "commit"
    
    id = factory.Sequence(lambda n: n)
    name = factory.Faker("name", locale="zh_CN")
    email = factory.LazyAttribute(lambda obj: f"{obj.name.lower().replace(' ', '.')}@example.com")

# test_factories.py
import pytest
from factories import UserFactory, SessionLocal
from database import User

class TestWithFactories:
    """使用 Factory Boy 的测试"""
    
    def test_create_single_user(self):
        """测试创建单个用户"""
        user = UserFactory()
        
        assert user.id is not None
        assert user.name is not None
        assert "@example.com" in user.email
        print(f"\n创建用户: {user.name} - {user.email}")
    
    def test_create_batch_users(self):
        """测试批量创建用户"""
        users = UserFactory.create_batch(5)
        
        assert len(users) == 5
        # 每个用户都有唯一的邮箱
        emails = [u.email for u in users]
        assert len(set(emails)) == 5
    
    def test_custom_attributes(self):
        """测试自定义属性"""
        user = UserFactory(name="特定名称", email="specific@example.com")
        
        assert user.name == "特定名称"
        assert user.email == "specific@example.com"
    
    def test_build_vs_create(self):
        """测试 build 和 create 的区别"""
        # build: 只创建对象，不保存到数据库
        user_built = UserFactory.build()
        assert user_built.id is None  # 没有 ID
        
        # create: 保存到数据库
        user_created = UserFactory()
        assert user_created.id is not None  # 有 ID

5.3 异步数据库测试

# async_database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, select

Base = declarative_base()

class Product(Base):
    __tablename__ = "products"
    
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    price = Column(Integer, nullable=False)

# test_async_db.py
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from async_database import Base, Product

TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest_asyncio.fixture
async def async_session():
    """异步数据库会话 fixture"""
    engine = create_async_engine(TEST_DATABASE_URL)
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    
    async with async_session() as session:
        yield session
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    
    await engine.dispose()

@pytest.mark.asyncio
class TestAsyncDatabase:
    """异步数据库测试"""
    
    async def test_create_product(self, async_session):
        """测试创建产品"""
        product = Product(name="测试产品", price=100)
        async_session.add(product)
        await async_session.commit()
        
        # 异步查询
        result = await async_session.execute(
            select(Product).where(Product.name == "测试产品")
        )
        db_product = result.scalar_one()
        
        assert db_product.price == 100
    
    async def test_update_product(self, async_session):
        """测试更新产品"""
        product = Product(name="可更新产品", price=50)
        async_session.add(product)
        await async_session.commit()
        
        # 更新价格
        product.price = 75
        await async_session.commit()
        
        # 验证更新
        result = await async_session.execute(
            select(Product).where(Product.id == product.id)
        )
        updated = result.scalar_one()
        assert updated.price == 75

---

6. 测试覆盖率

6.1 生成覆盖率报告

# 基础覆盖率测试
pytest --cov=my_project

# 生成详细报告
pytest --cov=my_project --cov-report=term-missing

# 生成 HTML 报告
pytest --cov=my_project --cov-report=html

# 生成 XML 报告（用于 CI/CD）
pytest --cov=my_project --cov-report=xml

# 设置覆盖率阈值
pytest --cov=my_project --cov-fail-under=80

6.2 覆盖率配置文件

# .coveragerc
[run]
source = my_project
omit = 
    */tests/*
    */venv/*
    */migrations/*
    */__pycache__/*
    setup.py

[report]
exclude_lines =
    pragma: no cover
    def __repr__
    raise NotImplementedError
    if __name__ == .__main__.:
    pass

show_missing = True
skip_covered = False

[html]
directory = htmlcov

6.3 分析覆盖率报告

# 运行测试并生成 HTML 报告
pytest --cov=my_project --cov-report=html

# 报告输出位置
# htmlcov/index.html

HTML 报告包含：

• 总体覆盖率百分比
• 每个文件的覆盖率
• 未覆盖的代码行（红色标记）
• 已覆盖的代码行（绿色标记）

---

避坑小贴士

1. 测试之间相互影响

# 错误：测试之间共享状态
users = []

def test_add_user():
    users.append({"name": "张三"})
    assert len(users) == 1

def test_another_user():
    # 这个测试可能会失败，因为 users 已经有数据了
    users.append({"name": "李四"})
    assert len(users) == 1  # 可能失败！

# 正确：使用 fixture 隔离状态
import pytest

@pytest.fixture
def user_list():
    return []

def test_add_user(user_list):
    user_list.append({"name": "张三"})
    assert len(user_list) == 1

def test_another_user(user_list):
    # 每个测试都有独立的 user_list
    user_list.append({"name": "李四"})
    assert len(user_list) == 1  # 一定成功

2. 忘记清理资源

# 错误：不清理资源
@pytest.fixture
def file_handle():
    f = open("test.txt", "w")
    return f  # 文件不会关闭！

# 正确：使用 yield 清理
@pytest.fixture
def file_handle():
    f = open("test.txt", "w")
    yield f
    f.close()  # 测试后清理

# 更好：使用上下文管理器
@pytest.fixture
def file_handle():
    with open("test.txt", "w") as f:
        yield f  # 自动关闭

3. 测试名称不清晰

# 错误：测试名不清楚做什么
def test1():
    assert add(1, 2) == 3

def test_func():
    assert multiply(2, 3) == 6

# 正确：描述性的测试名
def test_add_positive_numbers_returns_sum():
    """测试正数相加返回正确的和"""
    assert add(1, 2) == 3

def test_multiply_two_positive_numbers():
    """测试两个正数相乘"""
    assert multiply(2, 3) == 6

4. 一个测试验证太多东西

# 错误：测试太复杂
def test_user_system():
    # 创建用户
    user = create_user("张三")
    assert user.name == "张三"
    
    # 更新用户
    update_user(user.id, name="李四")
    assert get_user(user.id).name == "李四"
    
    # 删除用户
    delete_user(user.id)
    assert get_user(user.id) is None

# 正确：拆分测试
def test_create_user():
    user = create_user("张三")
    assert user.name == "张三"

def test_update_user():
    user = create_user("张三")
    update_user(user.id, name="李四")
    assert get_user(user.id).name == "李四"

def test_delete_user():
    user = create_user("张三")
    delete_user(user.id)
    assert get_user(user.id) is None

5. 忽略边界条件

# 错误：只测试正常情况
def test_divide():
    assert divide(10, 2) == 5

# 正确：测试边界条件
def test_divide_normal():
    assert divide(10, 2) == 5

def test_divide_by_zero():
    with pytest.raises(ZeroDivisionError):
        divide(10, 0)

def test_divide_negative():
    assert divide(-10, 2) == -5
    assert divide(10, -2) == -5
    assert divide(-10, -2) == 5

def test_divide_float():
    assert divide(5, 2) == 2.5

---

课后练习

练习1：为待办事项应用编写测试

# todo_app.py
from flask import Flask, request, jsonify

app = Flask(__name__)

todos = {}
next_id = 1

class TodoService:
    """待办事项服务"""
    
    @staticmethod
    def create(title, description=""):
        global next_id
        todo = {
            "id": next_id,
            "title": title,
            "description": description,
            "completed": False
        }
        todos[next_id] = todo
        next_id += 1
        return todo
    
    @staticmethod
    def get(todo_id):
        return todos.get(todo_id)
    
    @staticmethod
    def update(todo_id, **kwargs):
        if todo_id not in todos:
            return None
        todos[todo_id].update(kwargs)
        return todos[todo_id]
    
    @staticmethod
    def delete(todo_id):
        return todos.pop(todo_id, None)
    
    @staticmethod
    def list_all():
        return list(todos.values())
    
    @staticmethod
    def mark_completed(todo_id):
        return TodoService.update(todo_id, completed=True)
    
    @staticmethod
    def get_by_status(completed):
        return [t for t in todos.values() if t["completed"] == completed]

# 编写以下测试：
# 1. test_create_todo - 测试创建待办事项
# 2. test_get_todo - 测试获取单个待办事项
# 3. test_update_todo - 测试更新待办事项
# 4. test_delete_todo - 测试删除待办事项
# 5. test_mark_completed - 测试标记完成
# 6. test_filter_by_status - 测试按状态筛选
# 7. test_get_nonexistent - 测试获取不存在的待办事项

练习2：模拟外部 API 调用

# weather_service.py
import requests

class WeatherService:
    """天气服务"""
    
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.weather.com/v1"
    
    def get_current_weather(self, city):
        """获取当前天气"""
        response = requests.get(
            f"{self.base_url}/current",
            params={"city": city, "apikey": self.api_key}
        )
        response.raise_for_status()
        data = response.json()
        return {
            "temperature": data["temp"],
            "humidity": data["humidity"],
            "description": data["weather"][0]["description"]
        }
    
    def get_forecast(self, city, days=3):
        """获取天气预报"""
        response = requests.get(
            f"{self.base_url}/forecast",
            params={"city": city, "days": days, "apikey": self.api_key}
        )
        response.raise_for_status()
        return response.json()["forecast"]

# 编写测试：
# 1. 使用 @patch 模拟 requests.get
# 2. 测试正常返回天气数据
# 3. 测试 API 返回错误时抛出异常
# 4. 测试网络超时情况

练习3：集成测试与覆盖率

# calculator.py
class Calculator:
    """计算器类"""
    
    def add(self, a, b):
        return a + b
    
    def subtract(self, a, b):
        return a - b
    
    def multiply(self, a, b):
        return a * b
    
    def divide(self, a, b):
        if b == 0:
            raise ValueError("Cannot divide by zero")
        return a / b
    
    def power(self, base, exponent):
        return base ** exponent
    
    def sqrt(self, n):
        if n < 0:
            raise ValueError("Cannot calculate square root of negative number")
        return n ** 0.5
    
    def factorial(self, n):
        if n < 0:
            raise ValueError("Factorial not defined for negative numbers")
        if n == 0 or n == 1:
            return 1
        result = 1
        for i in range(2, n + 1):
            result *= i
        return result

# 任务：
# 1. 为 Calculator 类编写完整的单元测试
# 2. 确保覆盖所有方法
# 3. 测试正常情况和异常情况
# 4. 运行 pytest --cov=calculator --cov-report=html
# 5. 查看覆盖率报告，确保达到 100% 覆盖

---

下一篇预告

第22讲：Docker容器化部署

在下一讲中，我们将学习：

• Dockerfile 编写与镜像构建
• Docker Compose 多服务编排
• 多阶段构建优化镜像大小
• 镜像安全扫描与最佳实践

---

参考资源

1. pytest 官方文档
2. FastAPI 测试指南
3. Flask 测试文档
4. Factory Boy 文档

---

本讲内容到此结束，如有疑问欢迎在评论区留言讨论！