Python脚本集成测试数据库如何隔离

wen 实用脚本 2026-07-19 1

本文目录导读：

Python脚本集成测试数据库如何隔离

每个测试函数独立的数据库事务（最推荐）
使用临时数据库（推荐）
PostgreSQL的Schema隔离（适合大项目）
使用容器化数据库（生产级方案）
混合策略：事务 + 工厂模式
最佳实践建议
完整示例：最常用的pytest配置

在Python中做数据库集成测试时,隔离性是保证测试可靠性和并行性的关键，以下是几种主流的隔离策略，按推荐程度排序：

每个测试函数独立的数据库事务（最推荐）

使用pytest的fixture，在每个测试前后回滚事务：

import pytest
from myapp.database import get_db_session
@pytest.fixture
def db_session():
    """每个测试函数使用独立的事务，测试结束后回滚"""
    session = get_db_session()
    try:
        # 开始事务
        yield session
    finally:
        # 回滚所有更改，确保隔离
        session.rollback()
        session.close()
def test_create_user(db_session):
    user = User(name="test", email="test@example.com")
    db_session.add(user)
    db_session.commit()
    # 验证数据存在
    saved = db_session.query(User).filter_by(email="test@example.com").first()
    assert saved is not None
# 测试结束后，所有数据被回滚，不影响其他测试

使用临时数据库（推荐）

创建独立的临时数据库,测试完成后销毁：

import pytest
import tempfile
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
@pytest.fixture(scope="function")
def temp_db():
    """创建临时SQLite数据库"""
    # 创建临时文件
    db_fd, db_path = tempfile.mkstemp(suffix='.db')
    os.close(db_fd)
    # 创建数据库连接
    engine = create_engine(f'sqlite:///{db_path}')
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    # 清理
    session.close()
    os.unlink(db_path)
def test_something(temp_db):
    # 使用临时数据库测试
    pass

PostgreSQL的Schema隔离（适合大项目）

为每个测试创建独立的Schema：

import pytest
from sqlalchemy import create_engine
from sqlalchemy_utils import create_database, drop_database
import uuid
@pytest.fixture(scope="function")
def pg_schema():
    """为每个测试创建独立的PostgreSQL schema"""
    db_url = "postgresql://user:pass@localhost:5432/testdb"
    # 生成唯一的schema名
    schema_name = f"test_{uuid.uuid4().hex[:8]}"
    engine = create_engine(db_url)
    # 创建schema
    with engine.connect() as conn:
        conn.execute(f"CREATE SCHEMA IF NOT EXISTS {schema_name}")
        conn.execute(f"SET search_path TO {schema_name}")
        Base.metadata.create_all(conn)
        conn.commit()
    # 创建指向该schema的session
    session = sessionmaker(bind=engine)()
    yield session
    # 清理
    session.close()
    with engine.connect() as conn:
        conn.execute(f"DROP SCHEMA {schema_name} CASCADE")
        conn.commit()

使用容器化数据库（生产级方案）

使用testcontainers库启动独立的数据库容器：

import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine
@pytest.fixture(scope="session")
def postgres_container():
    """启动PostgreSQL容器"""
    with PostgresContainer("postgres:15") as postgres:
        yield postgres
@pytest.fixture
def db_session(postgres_container):
    """使用容器数据库"""
    engine = create_engine(postgres_container.get_connection_url())
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()
def test_with_real_db(db_session):
    # 测试代码
    pass

混合策略：事务 + 工厂模式

# conftest.py
import pytest
from factory import Factory, Faker
class UserFactory(Factory):
    class Meta:
        model = User
    name = Faker('name')
    email = Faker('email')
@pytest.fixture
def user_factory(db_session):
    """创建测试数据工厂"""
    def create_user(**kwargs):
        return UserFactory.create(session=db_session, **kwargs)
    return create_user
def test_user_creation(user_factory, db_session):
    user = user_factory()
    assert db_session.query(User).count() == 1

最佳实践建议

开发环境：使用事务回滚（方法1），速度快
CI/CD环境：使用临时数据库或容器（方法2或4），更可靠
大型项目：使用Schema隔离（方法3），支持并行测试
对性能敏感：使用共享数据库但scope="function"的事务

完整示例：最常用的pytest配置

# conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from myapp.models import Base
# 使用内存SQLite，速度最快
TEST_DATABASE_URL = "sqlite:///:memory:"
@pytest.fixture(scope="session")
def engine():
    """创建数据库引擎（会话级别）"""
    engine = create_engine(TEST_DATABASE_URL)
    Base.metadata.create_all(engine)
    return engine
@pytest.fixture
def db_session(engine):
    """每个测试方法独立的session，支持回滚"""
    connection = engine.connect()
    transaction = connection.begin()
    Session = sessionmaker(bind=connection)
    session = Session()
    yield session
    session.close()
    transaction.rollback()
    connection.close()
# 测试函数中使用
def test_user(db_session):
    user = User(name="Alice")
    db_session.add(user)
    db_session.commit()
    assert db_session.query(User).count() == 1

选择哪种方案取决于你的项目规模、测试速度和可靠性要求，对于大多数Python项目，方法1（事务回滚） 是最佳平衡点。