本节目标:把「应用与外部依赖之间」的那层测试搭起来——用可跑的 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 实例 |
| 下游 HTTP | respx/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 分层与插件 · 下一节:属性测试、性能回归与覆盖率门禁 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。