一、异步代码测试（适配异步接口 / 框架）

随着 FastAPI、Sanic 等异步框架普及，pytest 对异步代码的测试能力成为必备技能：

1. 基础异步测试（pytest-asyncio）

先安装依赖：pip install pytest-asyncio aiohttp

# conftest.py
import pytest

# 启用asyncio插件（全局生效）
pytest_plugins = ["pytest_asyncio"]

# 异步fixture（scope和同步fixture一致）
@pytest.fixture(scope="session")
async def async_api_client():
    """异步HTTP客户端（aiohttp）"""
    import aiohttp
    session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10))
    yield session
    # 异步后置操作：关闭会话
    await session.close()

# 异步测试用例（函数前加async）
@pytest.mark.asyncio  # 标记为异步用例
async def test_async_api(async_api_client):
    # 异步请求FastAPI接口
    resp = await async_api_client.get("http://127.0.0.1:8000/api/users")
    # 断言响应
    assert resp.status == 200
    resp_json = await resp.json()  # 异步解析JSON
    assert isinstance(resp_json["data"], list)

2. 异步数据库测试（以 asyncpg 为例）

import pytest
import asyncpg

# 异步数据库夹具
@pytest.fixture(scope="module")
async def async_db_conn():
    # 建立异步数据库连接
    conn = await asyncpg.connect(
        user="test",
        password="123456",
        database="test_db",
        host="127.0.0.1"
    )
    # 前置操作：创建测试表
    await conn.execute("""
        CREATE TABLE IF NOT EXISTS test_users (
            id SERIAL PRIMARY KEY,
            name VARCHAR(50) NOT NULL
        )
    """)
    yield conn
    # 后置操作：清理数据+关闭连接
    await conn.execute("DROP TABLE test_users")
    await conn.close()

# 测试异步数据库操作
@pytest.mark.asyncio
async def test_async_db(async_db_conn):
    # 插入数据
    await async_db_conn.execute("INSERT INTO test_users (name) VALUES ('test')")
    # 查询数据
    record = await async_db_conn.fetchrow("SELECT * FROM test_users WHERE name = 'test'")
    assert record["name"] == "test"

二、测试数据隔离与工厂模式（解决数据污染）

在批量测试中，测试数据相互污染是高频痛点，工厂模式 + 数据隔离能完美解决：

1. 测试数据工厂（factory-boy）

先安装依赖：pip install factory-boy faker（faker 用于生成模拟数据）

# factories/user_factory.py
import factory
from faker import Faker
from datetime import datetime

# 初始化faker（支持中文）
fake = Faker("zh_CN")

class UserFactory(factory.DictFactory):
    """用户数据工厂（生成模拟用户字典）"""
    # 动态生成唯一用户名
    username = factory.LazyAttribute(lambda x: fake.user_name() + str(datetime.now().microsecond))
    # 生成手机号
    phone = factory.LazyAttribute(lambda x: fake.phone_number())
    # 固定值+动态值结合
    email = factory.LazyAttribute(lambda obj: f"{obj.username}@test.com")
    age = factory.Faker("random_int", min=18, max=60)

# 测试用例中使用
import pytest

def test_create_user():
    # 生成单个用户数据
    user1 = UserFactory()
    user2 = UserFactory()
    # 每个用户的用户名都是唯一的，避免数据冲突
    assert user1["username"] != user2["username"]
    assert 18 <= user1["age"] <= 60
    print(f"生成的用户1：{user1}")

def test_create_batch_users():
    # 批量生成10个用户数据
    users = UserFactory.create_batch(10)
    assert len(users) == 10
    # 验证所有用户名唯一
    usernames = [u["username"] for u in users]
    assert len(usernames) == len(set(usernames))

2. 数据库数据隔离（事务回滚）

# conftest.py
import pytest
import psycopg2
from psycopg2 import extensions

@pytest.fixture(scope="function")
def db_conn_with_rollback():
    """数据库连接夹具：每个用例执行完回滚事务，隔离数据"""
    conn = psycopg2.connect(
        user="test", password="123456", database="test_db", host="127.0.0.1"
    )
    # 设置事务为手动提交
    conn.set_isolation_level(extensions.ISOLATION_LEVEL_AUTOCOMMIT)
    cur = conn.cursor()
    
    # 开启事务
    cur.execute("BEGIN")
    yield cur  # 传递游标给用例
    
    # 后置操作：回滚事务，清除所有测试数据
    cur.execute("ROLLBACK")
    cur.close()
    conn.close()

# 测试用例
def test_db_isolation(db_conn_with_rollback):
    # 插入测试数据
    db_conn_with_rollback.execute("INSERT INTO users (name) VALUES ('test_rollback')")
    # 查询验证插入成功
    db_conn_with_rollback.execute("SELECT * FROM users WHERE name = 'test_rollback'")
    assert db_conn_with_rollback.fetchone() is not None

# 另一个用例：看不到上一个用例插入的数据
def test_db_isolation_2(db_conn_with_rollback):
    db_conn_with_rollback.execute("SELECT * FROM users WHERE name = 'test_rollback'")
    assert db_conn_with_rollback.fetchone() is None

三、自定义 pytest 插件（封装通用能力）

当项目中通用逻辑较多时，封装成自定义插件可大幅提升复用性：

1. 简单自定义插件（本地插件）

# plugins/pytest_custom_plugin.py
import pytest
import logging

# 1. 自定义标记注册
def pytest_configure(config):
    """注册自定义标记，避免运行时报警告"""
    config.addinivalue_line(
        "markers", "api: 标记接口测试用例"
    )
    config.addinivalue_line(
        "markers", "db: 标记数据库测试用例"
    )

# 2. 自定义钩子：修改测试报告标题
def pytest_html_report_title(report):
    report.title = "XXX项目自动化测试报告"

# 3. 自定义fixture（插件内的fixture）
@pytest.fixture(scope="session")
def custom_logger():
    """自定义日志夹具"""
    logger = logging.getLogger("pytest_custom")
    logger.setLevel(logging.INFO)
    # 添加控制台处理器
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s"))
    logger.addHandler(handler)
    return logger

# 4. 自定义命令行参数
def pytest_addoption(parser):
    parser.addoption("--test-id", action="store", help="指定测试用例ID运行")

# 5. 用例收集钩子：按测试ID筛选
def pytest_collection_modifyitems(items, config):
    test_id = config.getoption("--test-id")
    if not test_id:
        return
    # 筛选包含指定ID的用例（假设用例名称包含test_id）
    filtered_items = [item for item in items if test_id in item.name]
    items[:] = filtered_items

2. 加载自定义插件

# conftest.py
# 方式1：直接导入
import sys
sys.path.append("plugins/")
from pytest_custom_plugin import *

# 方式2：通过pytest.ini配置
# pytest.ini中添加：
# [pytest]
# plugins = plugins/pytest_custom_plugin.py

3. 使用自定义插件能力

def test_use_custom_plugin(custom_logger):
    # 使用自定义日志夹具
    custom_logger.info("这是自定义插件的日志")
    assert True

# 运行时按测试ID筛选
# pytest -v --test-id=123 test_demo.py

四、精准调试与问题定位（高阶排障）

1. 用例失败时自动截图 / 保存请求日志

# conftest.py
import pytest
import os
from datetime import datetime

# 创建失败用例日志目录
os.makedirs("failed_cases", exist_ok=True)

@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
    """钩子函数：捕获用例执行结果"""
    # 获取原始报告
    outcome = yield
    report = outcome.get_result()
    
    # 仅在测试用例失败时执行
    if report.when == "call" and report.failed:
        # 1. 保存用例失败信息到文件
        fail_time = datetime.now().strftime("%Y%m%d_%H%M%S")
        fail_file = f"failed_cases/{item.name}_{fail_time}.txt"
        with open(fail_file, "w", encoding="utf-8") as f:
            f.write(f"用例名称：{item.name}\n")
            f.write(f"失败时间：{fail_time}\n")
            f.write(f"失败原因：{str(call.excinfo)}\n")
        
        # 2. 如果是UI测试，这里可添加自动截图逻辑
        # if hasattr(item, "page"):
        #     item.page.screenshot(path=f"failed_cases/{item.name}_{fail_time}.png")
        
        # 3. 保存接口请求日志（如果有）
        if hasattr(item, "request_log"):
            with open(f"failed_cases/{item.name}_{fail_time}_request.txt", "w", encoding="utf-8") as f:
                f.write(item.request_log)

2. 交互式调试（pytest-pdb）

# 用例失败时自动进入pdb调试模式
pytest -v --pdb test_demo.py

# 指定用例进入调试（断点）
pytest -v test_demo.py::test_debug -x --pdb

# 测试用例中手动加断点
def test_debug():
    a = 1
    b = 2
    import pdb; pdb.set_trace()  # 手动断点
    assert a + b == 3

3. 性能分析（pytest-benchmark）

先安装依赖：pip install pytest-benchmark

# 测试函数执行性能
def test_benchmark_sort(benchmark):
    # 测试排序性能
    result = benchmark(sorted, [9,3,1,7,5,8,2,4,6])
    assert result == [1,2,3,4,5,6,7,8,9]

# 运行命令：pytest -v --benchmark-autosave test_demo.py
# 生成性能报告，对比多次运行的耗时差异

五、测试左移：单元测试深度优化

1. 代码覆盖率（pytest-cov）

# 安装依赖
pip install pytest-cov

# 运行测试并生成覆盖率报告
pytest -v --cov=src/ --cov-report=html --cov-report=term test_case/

• --cov=src/：指定要统计覆盖率的源码目录
• --cov-report=html：生成 HTML 覆盖率报告（在 htmlcov 目录）
• --cov-report=term：终端显示覆盖率摘要

2. 单元测试 mock 进阶（patch 多层调用）

# src/utils.py
def get_external_data():
    """调用外部接口获取数据"""
    import requests
    return requests.get("https://api.external.com/data").json()

def process_data():
    """处理外部数据"""
    data = get_external_data()
    return [item["id"] for item in data["list"]]

# 测试用例
import pytest
from unittest.mock import patch
from src.utils import process_data

@patch("src.utils.get_external_data")  # patch多层调用的函数
def test_process_data(mock_get):
    # 模拟返回值
    mock_get.return_value = {
        "list": [{"id": 1}, {"id": 2}, {"id": 3}]
    }
    # 调用被测试函数
    result = process_data()
    # 断言结果
    assert result == [1,2,3]
    # 验证mock被调用
    mock_get.assert_called_once()

总结

1. 异步测试：借助pytest-asyncio适配异步接口 / 数据库，是现代 Python 框架测试的必备能力；
2. 数据隔离：用factory-boy生成动态测试数据，结合事务回滚实现数据隔离，解决批量测试数据污染问题；
3. 自定义插件：封装通用逻辑（日志、标记、筛选），提升项目复用性和规范性；
4. 调试与分析：掌握失败自动记录、pdb 调试、覆盖率 / 性能分析，能快速定位复杂问题。