pytest-postgresql 6.1.1

连接到现有的PostgreSQL数据库

一些项目正在使用已经运行的PostgreSQL服务器（例如在Docker实例上）。为了连接到它们，人们将使用 postgresql_noproc 配置。

postgresql_external = factories.postgresql('postgresql_noproc')

默认情况下，postgresql_noproc 配置将连接到使用 5432 端口的PostgreSQL实例。标准配置选项适用于它。

这些是适用于所有级别与 postgresql_noproc 配置的配置选项

为测试填充数据库

使用 SQLAlchemy

此示例演示了如何填充数据库并创建 SQLAlchemy 的 ORM 连接

以下示例是从 pyramid_fullauth 测试中简化后的 session fixture

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.pool import NullPool
from zope.sqlalchemy import register


@pytest.fixture
def db_session(postgresql):
    """Session for SQLAlchemy."""
    from pyramid_fullauth.models import Base

    connection = f'postgresql+psycopg2://{postgresql.info.user}:@{postgresql.info.host}:{postgresql.info.port}/{postgresql.info.dbname}'

    engine = create_engine(connection, echo=False, poolclass=NullPool)
    pyramid_basemodel.Session = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
    pyramid_basemodel.bind_engine(
        engine, pyramid_basemodel.Session, should_create=True, should_drop=True)

    yield pyramid_basemodel.Session

    transaction.commit()
    Base.metadata.drop_all(engine)


@pytest.fixture
def user(db_session):
    """Test user fixture."""
    from pyramid_fullauth.models import User
    from tests.tools import DEFAULT_USER

    new_user = User(**DEFAULT_USER)
    db_session.add(new_user)
    transaction.commit()
    return new_user


def test_remove_last_admin(db_session, user):
    """
    Sample test checks internal login, but shows usage in tests with SQLAlchemy
    """
    user = db_session.merge(user)
    user.is_admin = True
    transaction.commit()
    user = db_session.merge(user)

    with pytest.raises(AttributeError):
        user.is_admin = False

在 fixtures 之外维护数据库状态

使用 pytest-postgresql 数据库管理功能，维护数据库状态是可能的，并且似乎在测试的库中使用

为此导入 DatabaseJanitor 并使用其 init 和 drop 方法

import pytest
from pytest_postgresql.janitor import DatabaseJanitor

@pytest.fixture
def database(postgresql_proc):
    # variable definition

    janitor = DatabaseJanitor(
        postgresql_proc.user,
        postgresql_proc.host,
        postgresql_proc.port,
        "my_test_database",
        postgresql_proc.version,
        password="secret_password",
    )
    janitor.init()
    yield psycopg2.connect(
        dbname="my_test_database",
        user=postgresql_proc.user,
        password="secret_password",
        host=postgresql_proc.host,
        port=postgresql_proc.port,
    )
    janitor.drop()

或将其用作上下文管理器

import pytest
from pytest_postgresql.janitor import DatabaseJanitor

@pytest.fixture
def database(postgresql_proc):
    # variable definition

    with DatabaseJanitor(
        postgresql_proc.user,
        postgresql_proc.host,
        postgresql_proc.port,
        "my_test_database",
        postgresql_proc.version,
        password="secret_password",
    ):
        yield psycopg2.connect(
            dbname="my_test_database",
            user=postgresql_proc.user,
            password="secret_password",
            host=postgresql_proc.host,
            port=postgresql_proc.port,
        )

连接到 Postgresql（在 docker 中）

要连接到运行在 docker 中的 postgresql 并在上面运行测试，请使用 noproc fixtures。

docker run --name some-postgres -e POSTGRES_PASSWORD=mysecretpassword -d postgres

这将启动在 docker 容器中的 postgresql，但是使用本地安装的 postgresql 并没有太大区别。

在测试中，请确保所有测试都使用 postgresql_noproc fixture，如下所示

from pytest_postgresql import factories


postgresql_in_docker = factories.postgresql_noproc()
postgresql = factories.postgresql("postgresql_in_docker", dbname="test")


def test_postgres_docker(postgresql):
    """Run test."""
    cur = postgresql.cursor()
    cur.execute("CREATE TABLE test (id serial PRIMARY KEY, num integer, data varchar);")
    postgresql.commit()
    cur.close()

然后运行测试

pytest --postgresql-host=172.17.0.2 --postgresql-password=mysecretpassword

所有测试的基本数据库状态

如果您有多个需要共同初始化的测试，您可以定义一个 load 并将其传递到您的自定义 postgresql 进程 fixture

import pytest_postgresql.factories
def load_database(**kwargs):
    db_connection: connection = psycopg2.connect(**kwargs)
    with db_connection.cursor() as cur:
        cur.execute("CREATE TABLE stories (id serial PRIMARY KEY, name varchar);")
        cur.execute(
            "INSERT INTO stories (name) VALUES"
            "('Silmarillion'), ('Star Wars'), ('The Expanse'), ('Battlestar Galactica')"
        )
        db_connection.commit()

postgresql_proc = factories.postgresql_proc(
    load=[load_database],
)

postgresql = factories.postgresql(
    "postgresql_proc",
)

此过程将是进程 fixture 填充模板数据库，然后客户端 fixture 将自动使用它来从头开始创建测试数据库。快速、干净且没有悬空事务，这些事务可能被意外回滚。

相同的做法将适用于 noproces fixture，无论连接到的是在 docker 机器上运行还是远程或本地运行的已运行的 postgresql 实例。

def load_database(**kwargs):
    connection = f"postgresql+psycopg2://{kwargs['user']}:@{kwargs['host']}:{kwargs['port']}/{kwargs['dbname']}"
    engine = create_engine(connection)
    Base.metadata.create_all(engine)
    session = scoped_session(sessionmaker(bind=engine))
    # add things to session
    session.commit()

postgresql_proc = factories.postgresql_proc(load=[load_database])

postgresql = factories.postgresql('postgresql_proc') # still need to check if this is actually needed or not

@pytest.fixture
def dbsession(postgresql):
    connection = f'postgresql+psycopg2://{postgresql.info.user}:@{postgresql.info.host}:{postgresql.info.port}/{postgresql.info.dbname}'
    engine = create_engine(connection)

    session = scoped_session(sessionmaker(bind=engine))

    yield session
    # 'Base.metadata.drop_all(engine)' here specifically does not work. It is also not needed. If you leave out the session.close()
    # all the tests still run, but you get a warning/error at the end of the tests.
    session.close()