.. image:: https://raw.githubusercontent.com/ClearcodeHQ/pytest-postgresql/master/logo.png :width: 100px :height: 100px

pytest-postgresql

.. image:: https://img.shields.io/pypi/v/pytest-postgresql.svg :target: https://pypi.python.org/pypi/pytest-postgresql/ :alt: 最新PyPI版本

.. image:: https://img.shields.io/pypi/wheel/pytest-postgresql.svg :target: https://pypi.python.org/pypi/pytest-postgresql/ :alt: Wheel状态

.. image:: https://img.shields.io/pypi/pyversions/pytest-postgresql.svg :target: https://pypi.python.org/pypi/pytest-postgresql/ :alt: 支持的Python版本

.. image:: https://img.shields.io/pypi/l/pytest-postgresql.svg :target: https://pypi.python.org/pypi/pytest-postgresql/ :alt: 许可证

这是什么？

这是一个pytest插件，可以让你测试依赖于运行中的PostgreSQL数据库的代码。它允许你为PostgreSQL进程和客户端指定fixtures。

如何使用

.. warning::

已在PostgreSQL 10及以上版本上测试。更多细节请参见测试。

安装方法：

.. code-block:: sh

pip install pytest-postgresql

你还需要安装psycopg。请参阅其安装说明 <https://www.psycopg.org/psycopg3/docs/basic/install.html>。 请注意，此插件需要psycopg版本3。可以同时安装版本3和版本2，以供需要后者的库使用（参见这些说明 <https://www.psycopg.org/docs/install.html>）。

插件包含三个fixtures：

• postgresql - 这是一个功能作用域的客户端fixture。每次测试后，它会结束所有剩余的连接，并从PostgreSQL中删除测试数据库以确保可重复性。此fixture返回已连接的psycopg连接。

• postgresql_proc - 会话作用域的fixture，在首次使用时启动PostgreSQL实例，并在测试结束时停止。

• postgresql_noproc - 一个无进程fixture，连接到已运行的postgresql实例。例如在dockerized测试环境或提供postgresql服务的CI上。

只需将这些fixtures中的一个包含在你的测试fixture列表中即可。

如果需要，你还可以创建额外的postgresql客户端和进程fixtures：

.. code-block:: python

from pytest_postgresql import factories

postgresql_my_proc = factories.postgresql_proc(
    port=None, unixsocketdir='/var/run')
postgresql_my = factories.postgresql('postgresql_my_proc')

.. note::

每个PostgreSQL进程fixture都可以通过fixture工厂参数进行不同的配置。

示例测试

.. code-block:: python

def test_example_postgres(postgresql):
    """检查主要postgresql fixture。"""
    cur = postgresql.cursor()
    cur.execute("CREATE TABLE test (id serial PRIMARY KEY, num integer, data varchar);")
    postgresql.commit()
    cur.close()

如果你希望数据库fixture自动填充你的模式，有两种方法：

#. 客户端fixture特定 #. 进程fixture特定

两者都接受相同的可能加载器集：

• sql文件路径
• 加载函数导入路径（字符串）
• 实际加载函数

该函数将接收host、port、user、dbname和password关键字参数，并需要在内部执行与数据库的连接。但是，你将能够运行SQL文件，甚至以编程方式触发你拥有的数据库迁移。

客户端特定方法在每次测试时加载数据库

.. code-block:: python

from pathlib import Path
postgresql_my_with_schema = factories.postgresql(
    'postgresql_my_proc',
    load=[Path("schemafile.sql"), Path("otherschema.sql"), "import.path.to.function", "import.path.to:otherfunction", load_this]
)

.. warning::

以这种方式，数据库仍然会在每次测试后被删除。

进程fixture在每个测试会话中执行一次加载，并将数据加载到模板数据库中。 然后，客户端fixture在每次测试时从模板数据库创建测试数据库，这显著加快了测试速度。

.. code-block:: python

from pathlib import Path
postgresql_my_proc = factories.postgresql_proc(
    load=[Path("schemafile.sql"), Path("otherschema.sql"), "import.path.to.function", "import.path.to:otherfunction", load_this]
)

.. code-block:: sh

pytest --postgresql-populate-template=path.to.loading_function --postgresql-populate-template=path.to.other:loading_function --postgresql-populate-template=path/to/file.sql

示例中的loading_function将接收参数，并需要提交。

连接到已存在的postgresql数据库

一些项目使用已运行的postgresql服务器（例如在docker实例上）。 为了连接到它们，可以使用postgresql_noproc fixture。

.. code-block:: python

postgresql_external = factories.postgresql('postgresql_noproc')

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

以下是适用于所有级别的postgresql_noproc fixture的配置选项：

配置

你可以通过三种方式定义设置：fixture工厂参数、命令行选项和pytest.ini配置选项。 你可以选择喜欢的方式，但请记住这些设置按以下顺序处理：

* ``Fixture工厂参数``
* ``命令行选项``
* ``pytest.ini文件中的配置选项``

.. list-table:: 配置选项 :header-rows: 1

• • PostgreSQL选项
  • Fixture工厂参数
  • 命令行选项
  • pytest.ini选项
  • 无进程fixture
  • 默认值
• • 可执行文件路径
  • executable
  • --postgresql-exec
  • postgresql_exec
  • /usr/lib/postgresql/13/bin/pg_ctl
• • 主机
  • host
  • --postgresql-host
  • postgresql_host
  • 是
  • 127.0.0.1
• • 端口
  • port
  • --postgresql-port
  • postgresql_port
  • 是 (5432)
  • 随机
• • postgresql用户
  • user
  • --postgresql-user
  • postgresql_user
  • 是
  • postgres
• • 密码
  • password
  • --postgresql-password
  • postgresql_password
  • 是
• • 启动参数（额外的pg_ctl参数）
  • startparams
  • --postgresql-startparams
  • postgresql_startparams
  • -w
• • Postgres exe额外参数（通过pg_ctl的-o参数传递）
  • postgres_options
  • --postgresql-postgres-options
  • postgresql_postgres_options
• • unixsockets位置
  • unixsocket
  • --postgresql-unixsocketdir
  • postgresql_unixsocketdir
  • $TMPDIR
• • fixtures将创建的数据库名称
  • dbname
  • --postgresql-dbname
  • postgresql_dbname
  • 是，但使用xdist时会为每个工作进程添加索引，生成test0、test1等名称。
  • test
• • 默认模式，可以是sql文件或导入路径到加载函数（每个值的列表）
  • load
  • --postgresql-load
  • postgresql_load
  • 是
• • PostgreSQL连接选项
  • options
  • --postgresql-options
  • postgresql_options
  • 是

使用示例：

• 在你自己的fixture中作为参数传递

  .. code-block:: python

    postgresql_proc = factories.postgresql_proc(
        port=8888)
  

• 运行测试时使用--postgresql-port命令行选项

  .. code-block:: sh

    py.test tests --postgresql-port=8888
  

• 在pytest.ini文件中将端口指定为postgresql_port。

  要这样做，请在pytest.ini文件的[pytest]部分下添加如下行：

  .. code-block:: ini

    [pytest]
    postgresql_port = 8888
  

示例

为测试填充数据库

使用SQLAlchemy +++++++++++++++

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

以下是来自pyramid_fullauth <https://github.com/fizyk/pyramid_fullauth/>_测试的简化会话fixture示例：

.. code-block:: python

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):
    """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):
    """测试用户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):
    """
    示例测试检查内部登录，但展示了在测试中使用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

.. note::

查看`pyramid_fullauth的conftest文件 <https://github.com/fizyk/pyramid_fullauth/blob/2950e7f4a397b313aaf306d6d1a763ab7d8abf2b/tests/conftest.py#L35>`_中的原始代码。
根据你的需求，中间的代码可以在SQLAlchemy堆栈的情况下触发alembic迁移或任何其他代码

在fixtures外维护数据库状态

可以使用pytest-postgresql数据库管理功能在fixtures外维护数据库状态，这似乎在其他测试库中也有使用：

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

.. code-block:: python

import pytest
from pytest_postgresql.janitor import DatabaseJanitor

@pytest.fixture
def database(postgresql_proc):
    # 变量定义

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()

或者将其作为上下文管理器使用：

.. code-block:: python

import pytest
from pytest_postgresql.janitor import DatabaseJanitor

@pytest.fixture
def database(postgresql_proc):
    # 变量定义

    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,
        )

.. note::

DatabaseJanitor管理数据库的状态，但你需要自己创建在测试代码中使用的连接。

你可以选择传入一个已识别的postgresql ISOLATION_LEVEL以获得额外的控制。

.. note::

查看DatabaseJanitor在Python的warehouse测试代码中的使用 https://github.com/pypa/warehouse/blob/5d15bfe/tests/conftest.py#L127

连接到Postgresql（在Docker中）

要连接到Docker运行的postgresql并在其上运行测试，请使用noproc固件。

.. code-block:: sh

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

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

在测试中，确保所有测试都使用postgresql_noproc固件，如下所示：

.. code-block:: python

from pytest_postgresql import factories


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


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

然后运行测试：

.. code-block:: sh

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

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

如果你有几个需要共同初始化的测试，你可以定义一个load并将其传递给自定义的postgresql进程固件：

.. code-block:: python

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",
)

这种方式的工作原理是，进程固件将填充模板数据库，然后客户端固件会自动使用该模板数据库从头创建一个测试数据库。 快速、干净，并且没有可能被意外回滚的悬挂事务。

同样的方法也适用于noproces固件，在连接到已经运行的postgresql实例时，无论它是在Docker机器上还是在远程或本地运行。

使用SQLAlchemy初始化基本数据库状态 +++++++++++++++++++++++++++++++++++++++++++++++++++

如何使用SQLAlchemy进行共同初始化：

.. code-block:: python

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))
    # 向session添加内容
    session.commit()

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

postgresql = factories.postgresql('postgresql_proc') # 仍需检查这是否实际需要

@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)'。也不需要。如果你省略session.close()，
    # 所有测试仍然会运行，但在测试结束时会收到警告/错误。
    session.close()

发布

首先安装pipenv和--dev依赖，然后运行：

.. code-block:: sh

pipenv run tbump [NEW_VERSION]