从手写初始化到 pytest fixture：让 Python 测试既干净、可复用，又能驾驭异步并发

Python 之所以迷人，不只是因为语法简洁，也因为它拥有一套成熟、开放、温暖的工程生态：Web 开发有 Django、Flask、FastAPI，数据分析有 NumPy、Pandas，AI 领域有 PyTorch、TensorFlow，自动化脚本、测试工具、运维平台也都能用 Python 快速落地。

但在真实项目中，决定一个 Python 工程能不能长期健康发展的，往往不是“你会不会写功能”，而是“你能不能放心修改功能”。而测试，正是这种放心感的来源。

今天我们聚焦一个非常实用的问题：

什么是 fixture？为什么它比“到处手写初始化代码”更强？当我们测试临时数据库、假用户、认证 token，甚至异步并发消费器时，fixture 应该如何设计？

pytest 官方文档把 fixtures 描述为一种可为测试提供固定基线的机制，让测试能够可靠、可重复地执行；同时它强调 fixture 具备显式、模块化、可扩展等优势，并能安全管理清理逻辑。(pytest 文档)

---

一、问题从哪里来：测试里的“复制粘贴初始化地狱”

假设我们正在开发一个订单系统，多个测试都需要：

• 一个临时数据库；
• 一个假用户；
• 一个认证 token；
• 一个已经登录的测试客户端。

初学者很容易写出这样的测试：

def test_create_order():
    db = create_temp_db()
    user = create_fake_user(db)
    token = create_token(user)
    client = TestClient(token=token)

    response = client.post("/orders", json={"sku": "BOOK", "count": 1})

    assert response.status_code == 201

    db.close()


def test_query_order():
    db = create_temp_db()
    user = create_fake_user(db)
    token = create_token(user)
    client = TestClient(token=token)

    response = client.get("/orders")

    assert response.status_code == 200

    db.close()

这段代码看起来没问题，但它有几个隐患。

第一，初始化逻辑重复。以后创建用户字段变了，你要改几十个测试。

第二，清理逻辑容易漏。中间如果断言失败，db.close() 可能不会执行。

第三，测试意图被噪音淹没。读者本来只想知道“这个测试验证什么”，却被数据库、用户、token 初始化细节打断。

第四，依赖关系散落各处。到底 token 依赖 user，user 依赖 db，client 依赖 token，这些关系没有被清楚表达。

fixture 解决的，正是这些工程化问题。

---

二、fixture 是什么：把 Arrange 变成可复用的测试资产

在测试中，我们常说有四个阶段：

Arrange：准备环境和数据
Act：执行被测行为
Assert：验证结果
Cleanup：清理资源

pytest 文档也用类似结构解释测试行为：准备、执行、断言、清理。(pytest 文档)

fixture 本质上就是把 Arrange 和 Cleanup 从测试函数里抽出来，变成可声明、可复用、可组合的组件。

一个最简单的 fixture：

import pytest

@pytest.fixture
def fake_user():
    return {
        "id": 1,
        "name": "Alice",
        "role": "customer",
    }


def test_user_name(fake_user):
    assert fake_user["name"] == "Alice"

你会发现，测试函数并没有手动调用 fake_user()，而是把 fake_user 写成参数。pytest 会根据参数名自动找到对应 fixture，并把返回值注入进来。

这就是 fixture 最重要的思想之一：

测试声明自己需要什么，而不是自己到处创建什么。

---

三、从临时数据库、假用户到认证 token：fixture 的组合能力

真实项目中，fixture 最强的地方不是“少写几行代码”，而是它可以表达依赖关系。

# conftest.py
import pytest
from myapp.db import create_test_db
from myapp.auth import create_token
from myapp.testing import TestClient


@pytest.fixture
def db():
    database = create_test_db()
    yield database
    database.drop_all()
    database.close()


@pytest.fixture
def fake_user(db):
    user = db.users.insert({
        "name": "Alice",
        "email": "alice@example.com",
        "role": "customer",
    })
    return user


@pytest.fixture
def auth_token(fake_user):
    return create_token(user_id=fake_user.id)


@pytest.fixture
def client(db, auth_token):
    return TestClient(database=db, token=auth_token)

测试代码立刻清爽很多：

def test_create_order(client):
    response = client.post("/orders", json={
        "sku": "BOOK-001",
        "count": 1,
    })

    assert response.status_code == 201
    assert response.json()["status"] == "created"


def test_query_orders(client):
    response = client.get("/orders")

    assert response.status_code == 200
    assert isinstance(response.json()["items"], list)

这里的依赖关系非常清楚：

db
 |
fake_user
 |
auth_token
 |
client

如果你想换一个管理员用户，也不必复制整套初始化代码。

@pytest.fixture
def admin_user(db):
    return db.users.insert({
        "name": "Root",
        "email": "root@example.com",
        "role": "admin",
    })


@pytest.fixture
def admin_token(admin_user):
    return create_token(user_id=admin_user.id)


@pytest.fixture
def admin_client(db, admin_token):
    return TestClient(database=db, token=admin_token)

测试也自然表达了业务语义：

def test_admin_can_delete_order(admin_client):
    response = admin_client.delete("/orders/1001")

    assert response.status_code == 204

这就是 fixture 比“手写初始化代码”强的地方：
它把测试依赖变成了可命名、可组合、可维护的结构。

---

四、yield fixture：清理逻辑应该和创建逻辑放在一起

很多测试资源都需要清理，比如临时数据库、临时文件、Redis key、消息队列 topic、mock server。

错误写法是让每个测试自己清理：

def test_something():
    db = create_test_db()
    # ...
    db.close()

更好的写法是用 yield fixture：

@pytest.fixture
def db():
    database = create_test_db()
    try:
        yield database
    finally:
        database.drop_all()
        database.close()

测试只关心使用：

def test_create_user(db):
    user = db.users.insert({"name": "Alice"})
    assert user.id is not None

无论测试成功还是失败，fixture 的清理逻辑都会执行。pytest 官方文档也强调，fixture 可以安全管理 teardown 逻辑，不需要测试作者手动维护复杂清理顺序。(pytest 文档)

这对工程质量非常关键。因为测试最怕的不是失败，而是失败后污染环境，导致后面的测试也变得诡异。

---

五、fixture scope：不是所有资源都该每次重建

pytest fixture 支持不同作用域。官方文档提到 fixture 可以在 function、class、module 或 session 等不同范围内复用。(pytest 文档)

常见选择如下：

scope	生命周期	适合场景
function	每个测试函数一次	默认选择，隔离性最好
module	每个测试文件一次	创建成本较高但可共享
session	整个测试会话一次	全局 mock server、测试容器
class	每个测试类一次	类组织风格的测试

比如数据库连接池可以是 session 级别，但每个测试的数据事务最好是 function 级别。

@pytest.fixture(scope="session")
def db_engine():
    engine = create_engine_for_test()
    yield engine
    engine.dispose()


@pytest.fixture
def db_session(db_engine):
    session = db_engine.create_session()
    transaction = session.begin()

    yield session

    transaction.rollback()
    session.close()

这样既避免每个测试都重复创建昂贵连接，又能保证每个测试的数据隔离。

这是 fixture 设计中的核心平衡：

重资源可以共享，脏数据必须隔离。

---

六、factory fixture：别让 fixture 变成万能大礼包

有时候，一个测试需要创建多个不同用户。不要写一堆 fixture：

@pytest.fixture
def user_a(): ...
@pytest.fixture
def user_b(): ...
@pytest.fixture
def vip_user(): ...
@pytest.fixture
def banned_user(): ...

更好的方式是提供一个工厂 fixture：

@pytest.fixture
def user_factory(db):
    def create_user(
        name="Alice",
        role="customer",
        email=None,
    ):
        return db.users.insert({
            "name": name,
            "email": email or f"{name.lower()}@example.com",
            "role": role,
        })

    return create_user

使用时：

def test_vip_user_gets_discount(client, user_factory):
    user = user_factory(name="Bob", role="vip")

    response = client.get(f"/discounts?user_id={user.id}")

    assert response.json()["discount"] > 0

factory fixture 的好处是：
默认值让简单测试很轻松，参数又让复杂测试保持灵活。

这比创建十几个高度具体的 fixture 更可维护。

---

七、fixture 太重，会带来什么问题？

fixture 是好东西，但设计得太重，会反过来伤害测试。

典型问题有四类。

1. 测试变慢

如果一个 client fixture 默认启动数据库、Redis、消息队列、浏览器、外部 mock server，那么每个简单单元测试都会背上沉重成本。

@pytest.fixture
def client():
    db = start_database()
    redis = start_redis()
    mq = start_message_queue()
    browser = start_browser()
    return FullStackClient(db, redis, mq, browser)

这会让测试套件越来越慢，最后团队不愿意跑测试。

2. 测试意图不清晰

当一个 fixture 做了太多事情，测试读起来就像魔法：

def test_checkout(client):
    response = client.post("/checkout")
    assert response.status_code == 200

问题是：用户是谁？购物车里有什么？库存是否足够？支付是否 mock？读者完全不知道。

3. 隐式耦合增加

如果很多测试都依赖同一个巨大 fixture，改它一次可能影响几百个测试。

这类 fixture 表面上复用率高，实际上是全局耦合点。

4. 失败定位困难

当测试失败时，你不知道是业务逻辑错了，还是 fixture 里某个隐藏初始化步骤错了。

所以 fixture 的最佳实践是：

小而清晰
显式命名
单一职责
默认简单
按需组合
避免全局魔法

一个好的 fixture 应该像乐高积木，而不是一辆焊死的工程车。

---

八、进阶案例：测试异步并发消费器

现在进入更高级的场景：我们要测试一个异步消费器。

需求是：

1. 从队列中消费订单；
2. 调用处理函数；
3. 验证处理结果；
4. 能正确响应取消；
5. 超时时不会让测试卡死。

先写一个简单消费者：

import asyncio

class OrderConsumer:
    def __init__(self, queue, handler):
        self.queue = queue
        self.handler = handler

    async def run(self):
        try:
            while True:
                order = await self.queue.get()
                try:
                    await self.handler(order)
                finally:
                    self.queue.task_done()
        except asyncio.CancelledError:
            # 做必要清理，然后继续抛出取消异常
            raise

pytest 本身测试普通函数很自然，但异步测试需要插件支持。pytest-asyncio 官方文档说明，它是 pytest 的 asyncio 插件，支持把协程作为测试函数，从而可以在测试中直接 await。(pytest-asyncio.readthedocs.io)

安装后可以这样写：

pip install pytest-asyncio

异步测试示例：

import pytest
import asyncio

@pytest.mark.asyncio
async def test_consumer_processes_orders():
    queue = asyncio.Queue()
    processed = []

    async def handler(order):
        processed.append(order)

    consumer = OrderConsumer(queue, handler)
    task = asyncio.create_task(consumer.run())

    await queue.put({"id": 1})
    await queue.put({"id": 2})

    await queue.join()

    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task

    assert processed == [{"id": 1}, {"id": 2}]

这个测试验证了结果，也正确取消了后台任务。

注意：不要创建后台任务后不保存引用。Python 官方 asyncio 文档提醒，事件循环只对 task 保持弱引用；可靠的后台任务应该保存引用。(Python documentation)

---

九、验证超时：测试不能无限等待

异步测试最危险的失败方式不是红，而是永远不结束。

所以测试并发消费者时，要给关键等待加超时。

@pytest.mark.asyncio
async def test_consumer_timeout_when_handler_hangs():
    queue = asyncio.Queue()

    async def hanging_handler(order):
        await asyncio.sleep(999)

    consumer = OrderConsumer(queue, hanging_handler)
    task = asyncio.create_task(consumer.run())

    await queue.put({"id": 1})

    with pytest.raises(TimeoutError):
        async with asyncio.timeout(0.05):
            await queue.join()

    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task

asyncio.timeout() 可以限制等待时间；超时后会取消当前任务，并把内部的 CancelledError 转换为可捕获的 TimeoutError。(Python documentation)

这段测试的重点不是语法，而是安全性：
任何等待外部事件的异步测试，都应该有退出路径。

---

十、验证取消：不要吞掉 CancelledError

异步消费者必须能被取消。错误写法是这样：

async def run(self):
    try:
        while True:
            order = await self.queue.get()
            await self.handler(order)
            self.queue.task_done()
    except Exception:
        pass

这段代码的问题是，它可能掩盖真实异常；更糟的是，如果错误地捕获取消异常，消费者就可能无法正常退出。

推荐写法：

async def run(self):
    try:
        while True:
            order = await self.queue.get()
            try:
                await self.handler(order)
            finally:
                self.queue.task_done()
    except asyncio.CancelledError:
        # 记录日志或释放资源
        raise

Python 官方文档建议协程使用 try/finally 做清理；如果显式捕获 CancelledError，通常应在清理完成后继续传播。(Python documentation)

对应测试：

@pytest.mark.asyncio
async def test_consumer_can_be_cancelled():
    queue = asyncio.Queue()

    async def handler(order):
        await asyncio.sleep(1)

    consumer = OrderConsumer(queue, handler)
    task = asyncio.create_task(consumer.run())

    await asyncio.sleep(0)

    task.cancel()

    with pytest.raises(asyncio.CancelledError):
        await task

    assert task.cancelled()

---

十一、为什么异步测试最难的不是语法，而是确定性？

很多人学异步测试，第一关是语法：

@pytest.mark.asyncio
async def test_xxx():
    result = await do_something()
    assert result == expected

但真正困难的不是 async def 和 await，而是 确定性。

并发程序有调度顺序。今天任务 A 先执行，明天可能任务 B 先执行。你写的测试如果依赖“刚好 sleep 了 0.1 秒之后某件事应该发生”，它就会变成脆弱测试。

脆弱写法：

await queue.put(order)
await asyncio.sleep(0.1)
assert processed == [order]

更稳定的写法是使用明确同步点：

@pytest.mark.asyncio
async def test_consumer_with_event_sync():
    queue = asyncio.Queue()
    processed = []
    done = asyncio.Event()

    async def handler(order):
        processed.append(order)
        done.set()

    consumer = OrderConsumer(queue, handler)
    task = asyncio.create_task(consumer.run())

    await queue.put({"id": 1})

    async with asyncio.timeout(1):
        await done.wait()

    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task

    assert processed == [{"id": 1}]

这里的 Event 就是确定性同步点。测试不再猜“睡多久够”，而是等待明确事件发生。

异步测试的最佳实践可以总结为：

少用 sleep 猜时间
多用 Event / Queue.join / Future 建立同步点
所有等待都加 timeout
后台 task 必须 cancel 并 await
不要吞 CancelledError
测试结果，也测试退出路径

---

十二、把 fixture 和异步测试结合起来

我们可以把异步消费者测试里的公共资源也做成 fixture。

@pytest.fixture
def order_queue():
    return asyncio.Queue()


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


@pytest.fixture
def order_handler(processed_orders):
    async def handler(order):
        processed_orders.append(order)
    return handler


@pytest.fixture
def consumer(order_queue, order_handler):
    return OrderConsumer(order_queue, order_handler)

测试变成：

@pytest.mark.asyncio
async def test_consumer_processes_one_order(
    order_queue,
    processed_orders,
    consumer,
):
    task = asyncio.create_task(consumer.run())

    await order_queue.put({"id": 1})
    await order_queue.join()

    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task

    assert processed_orders == [{"id": 1}]

如果需要更安全，还可以提供专门管理 task 生命周期的 async fixture。项目规模越大，越应该把“启动、取消、清理”的规则集中起来。

---

十三、实战中的一套分层建议

我通常会这样组织测试 fixture：

tests/
  conftest.py                 # 通用基础 fixture
  test_orders.py              # 订单接口测试
  test_auth.py                # 认证测试
  test_consumer.py            # 异步消费者测试

conftest.py 中放通用资源：

@pytest.fixture
def db_session():
    ...


@pytest.fixture
def user_factory(db_session):
    ...


@pytest.fixture
def token_factory():
    ...


@pytest.fixture
def client(db_session):
    ...

具体业务测试里再定义局部 fixture：

@pytest.fixture
def paid_order(user_factory):
    user = user_factory(role="customer")
    return {
        "user_id": user.id,
        "status": "paid",
        "sku": "BOOK-001",
    }

不要一开始就设计一个“万能测试世界”。fixture 应该从重复中生长出来，而不是凭空抽象出来。

---

十四、总结：fixture 是测试工程化的入口

fixture 的强大，不在于它让你少写几行初始化代码，而在于它改变了测试设计方式。

它让测试从：

我需要什么，就在这里手写什么

变成：

我声明需要什么，由测试系统组合出来

它让临时数据库、假用户、认证 token、测试客户端、异步队列、后台任务都能被清晰管理。

但也要记住：fixture 不是越多越好，也不是越大越好。太重的 fixture 会让测试变慢、变隐晦、变脆弱。优秀的 fixture 应该轻、准、清晰、有边界。

而异步测试提醒我们另一件事：高级测试能力不只是会写 await，而是能让并发行为变得可验证、可取消、可超时、可重复。

这也是 Python 工程实践最动人的地方：
它让初学者能快速开始，也给资深工程师留下足够深的空间去打磨质量、稳定性和长期可维护性。

你在项目里有没有遇到过“fixture 越写越重”或者“异步测试偶发失败”的问题？欢迎在评论区分享你的案例。真正好的 Python教程，不应该只教语法，也应该一起讨论这些真实工程里的取舍。