《Python编程实战》4.2 集成测试与容器化测试环境

单元测试之外的第二层防线:先划清单元与集成的边界,再用 SQLite + fixture 在本机搭出可跑的等价数据层,讲透外层事务加 SAVEPOINT 的用例隔离,最后给出 testcontainers 真实容器环境的配置并明确标注未实测。

本节目标:把「应用与外部依赖之间」的那层测试搭起来——用可跑的 SQLite 环境覆盖数据层,用事务回滚保证用例隔离,并知道何时该升级到真实容器。
适用版本:Python 3.12+(实测 3.14.6);SQLAlchemy 2.1.4

4.2 集成测试与容器化测试环境

上一节(pytest 工程化 )搭好了测试骨架,但跑的几乎都是纯逻辑。真实事故很少出在纯逻辑里,而多半出在「应用与数据库/缓存/外部服务之间」的边界上:SQL 写错、事务边界搞反、序列化字段对不上。这类问题只有集成测试能拦。

本节的实测项目用 SQLAlchemy 2.1.4 + SQLite 搭了一个最小数据层。本机没有 Docker,也没有安装 testcontainers,所以容器部分只给配置并明确标注未实测;可本地跑通的等价方案全部实测。

4.2.1 单元测试与集成测试的边界

先划边界,否则「什么该用集成测试」永远说不清:

维度单元测试集成测试
被测对象一个函数/方法两个及以上组件协作
外部依赖全部 mock真实(或高保真替身)
速度毫秒级百毫秒到秒级
失败含义逻辑错接线错、契约错、SQL 错
典型缺陷边界条件、分支遗漏事务边界、字段映射、方言差异

判断标准只有一条:如果把外部依赖换成 mock,这个 bug 还能被抓住吗? 能,就是单元测试;不能,就必须上集成测试。

4.2.2 集成测试要面对的四类外部依赖

一个典型后端服务的集成测试,绕不开这四类:

依赖单元测试里的替身集成测试里的真身
关系数据库Mock(spec=Session)真实数据库 + 迁移后的 schema
缓存/队列fakeredis真实 Redis 实例
下游 HTTPrespx/httpx.MockTransport契约桩或真实沙箱
消息中间件内存队列真实 broker

原则:越靠近数据一致性的依赖,越要用真身。数据库排第一,因为事务、约束、隔离级别这些行为,mock 根本模拟不出来。

4.2.3 用 SQLite 搭一个等价数据层

被测的数据访问层很小,但够真实——一张 users 表、唯一索引、flush 语义:

# app/db.py
from sqlalchemy import String, create_engine, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(100))


class UserRepo:
    def __init__(self, session: Session) -> None:
        self.session = session

    def create(self, email: str, name: str) -> User:
        user = User(email=email, name=name)
        self.session.add(user)
        self.session.flush()
        return user

    def get_by_email(self, email: str) -> User | None:
        return self.session.scalar(select(User).where(User.email == email))

    def count(self) -> int:
        return self.session.scalar(select(func.count()).select_from(User)) or 0

集成测试的 fixture 分三层,对应三种生命周期:

# tests/integration/conftest.py
import pytest
from sqlalchemy.orm import Session

from app.db import Base, UserRepo, make_engine


@pytest.fixture(scope="session")
def engine(tmp_path_factory):
    """一次启动、会话共享的数据库,等价于容器 fixture 的位置。"""
    db_path = tmp_path_factory.mktemp("db") / "test.db"
    eng = make_engine(f"sqlite+pysqlite:///{db_path}")
    Base.metadata.create_all(eng)
    yield eng
    eng.dispose()


@pytest.fixture(scope="session")
def connection(engine):
    with engine.connect() as conn:
        yield conn


@pytest.fixture
def repo(session):
    return UserRepo(session)

engine 是 session 级——建库、建表只做一次;connection 也复用同一条连接;session 是 function 级,每个测试一套独立事务。为什么用临时文件而不是 :memory::内存库随连接销毁,而 SAVEPOINT 隔离需要连接在多个测试间存活,文件库更稳。

4.2.4 用例隔离:外层事务 + SAVEPOINT

集成测试最大的坑是用例互相污染:前一个测试写的行留在库里,后一个测试的 count() 就莫名其妙不是 0。标准解法是「每个测试包一层事务,测完回滚」,但直接这么写会踩雷。先看错误示范:

@pytest.fixture
def session(engine):
    conn = engine.connect()
    trans = conn.begin()
    sess = Session(bind=conn)
    yield sess
    sess.close()
    trans.rollback()      # 天真写法
    conn.close()

跑一个会触发唯一约束冲突的用例(两次插入同一 email),teardown 直接炸:

ERROR at teardown of test_unique_email_conflict
    trans.rollback()
E   sqlalchemy.exc.SAWarning: transaction already deassociated from connection

原因:flush() 抛出 IntegrityError 时,SQLAlchemy 会自动回滚 Session 的事务,外层那个 trans 随即失效;再对它调 rollback() 就会报警告——而本项目配了 filterwarnings = ["error"],警告直接升级为错误。这不是配置太严,而是配置帮我们抓到了一个真实的资源管理缺陷。

正确写法用 join_transaction_mode="create_savepoint",让 Session 以 SAVEPOINT 参与外层事务:

@pytest.fixture
def session(connection):
    """每个测试套一层 SAVEPOINT,测完回滚,实现用例间隔离。"""
    trans = connection.begin()
    sess = Session(bind=connection, join_transaction_mode="create_savepoint")
    yield sess
    sess.close()
    trans.rollback()

这样即使某个用例的 flush 失败,回滚的也只是它自己的 SAVEPOINT,外层事务完好,teardown 干净利落。三个用例实测全绿:

import pytest
from app.db import UserRepo

pytestmark = pytest.mark.integration


def test_create_and_read(repo: UserRepo):
    repo.create("a@example.com", "Alice")
    repo.session.flush()
    found = repo.get_by_email("a@example.com")
    assert found is not None and found.name == "Alice"


def test_rollback_isolates_tests(repo: UserRepo):
    # 上一个测试写入的行不会出现在这里:外层事务已回滚
    assert repo.count() == 0


def test_unique_email_conflict(repo: UserRepo):
    from sqlalchemy.exc import IntegrityError

    repo.create("dup@example.com", "First")
    repo.session.flush()
    with pytest.raises(IntegrityError):
        repo.create("dup@example.com", "Second")
        repo.session.flush()
tests/integration/test_repo.py ...                                       [100%]
============================== 3 passed in 0.62s ===============================

test_rollback_isolates_tests 断言 count() == 0 却依然通过,正是隔离生效的铁证:test_create_and_read 明明插了一行,却因回滚从未真正落库。

4.2.5 testcontainers:真实容器环境(本机未实测)

SQLite 快、零依赖,但它和 PostgreSQL 之间隔着方言、扩展、隔离级别三道墙。当被测逻辑用到 PG 专属能力(JSONB、ON CONFLICT、SERIALIZABLE、pg_trgm)时,SQLite 的「通过」是假阳性。此时应换 testcontainers——以下代码本机没有 Docker,未实测,仅作配置示意:

# 示意:本机未安装 Docker / testcontainers,未执行
import pytest
from sqlalchemy import create_engine
from testcontainers.postgres import PostgresContainer


@pytest.fixture(scope="session")
def postgres():
    # 会话级启动一个真实 PG 容器,测试结束自动销毁
    with PostgresContainer("postgres:16-alpine") as pg:
        yield pg


@pytest.fixture(scope="session")
def engine(postgres):
    return create_engine(postgres.get_connection_url())

PostgresContainer 在会话级启动容器、暴露随机宿主机端口,get_connection_url() 返回一个可直接喂给 create_engine 的 URL。它与 4.2.3 的 SQLite engine fixture 位置完全对应——这也正是前面把 engine 抽成 session 级的原因:换后端时只改这一个 fixture,测试体一行不动。

对应地,安装依赖(未实测):

uv add --dev testcontainers[postgres]

4.2.6 本地替身与容器的取舍

两套方案不是二选一,而是分层:

场景用 SQLite 替身用 testcontainers
日常本地跑全量用例是(秒级反馈)否(启动慢)
PR 门禁是可选(跑关键子集)
发布前夜 / nightly否是(贴近生产)
依赖 PG 专属特性不可必须

务实做法:本地与 PR 用 SQLite 替身保速度,nightly 用容器保真实度,两套共用同一批测试体,只切换 fixture。这就是把后端差异收敛到一个 fixture 的回报。

4.2.7 用标记分层,接进 CI

上一节注册的 integration 标记在这里发挥作用。本地快速循环只跑单元测试:

pytest -m "not integration"     # 秒级,开发时高频跑
pytest -m integration           # 需要外部资源,CI 单独一档

CI 里建议分成两个 job:单元测试每次 push 都跑,集成测试在容器就绪后跑,失败时能一眼看出是「逻辑坏了」还是「环境/接线坏了」。数据层之外,缓存用 fakeredis、下游 HTTP 用 httpx.MockTransport 做契约桩,思路与本节一致——能真跑的用真身,不能的用高保真替身,且永远如实标注哪个是真跑的。

小结

  • 判断要不要集成测试只有一条标准:mock 掉外部依赖后这个 bug 还抓得住吗。
  • 数据一致性相关的依赖必须用真身,mock 模拟不出事务、约束、隔离级别。
  • 用 SQLite + tmp_path_factory 搭等价环境,把 engine 抽成 session 级,换后端只改一个 fixture。
  • 用例隔离靠「外层事务 + join_transaction_mode="create_savepoint"」;天真的 conn.begin() 写法会在 IntegrityError 后炸出 SAWarning。
  • testcontainers 提供真实容器环境,本地未实测,仅在依赖数据库专属特性时启用。

测试能跑、能隔离了,但「测多少才够、性能有没有退化」仍是两笔糊涂账。下一节用 hypothesis 补输入覆盖、用 pytest-benchmark 守住性能、用覆盖率门禁卡住底线。

延伸阅读:Python 测试与质量工程 、Python 数据库与 ORM 。

阅读导航:上一节:pytest 工程化:fixture 分层与插件 · 下一节:属性测试、性能回归与覆盖率门禁 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

  1. 《Python高级编程》目录
  2. 《Python高级编程》11.3 PEP 流程与版本迁移策略
  3. 《Python高级编程》11.2 嵌入式与自由线程运行时