pytest-postgresql 5.1.0

Connecting to already existing postgresql database

Some projects are using already running postgresql servers (ie on docker instances). In order to connect to them, one would be using the postgresql_noproc fixture.

postgresql_external = factories.postgresql('postgresql_noproc')

By default the postgresql_noproc fixture would connect to postgresql instance using 5432 port. Standard configuration options apply to it.

These are the configuration options that are working on all levels with the postgresql_noproc fixture:

Populating database for tests

With SQLAlchemy

This example shows how to populate database and create an SQLAlchemy’s ORM connection:

Sample below is simplified session fixture from pyramid_fullauth tests:

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

Maintaining database state outside of the fixtures

It is possible and appears it’s used in other libraries for tests, to maintain database state with the use of the pytest-postgresql database managing functionality:

For this import DatabaseJanitor and use its init and drop methods:

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

or use it as a context manager:

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

Connecting to Postgresql (in a docker)

To connect to a docker run postgresql and run test on it, use noproc fixtures.

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

This will start postgresql in a docker container, however using a postgresql installed locally is not much different.

In tests, make sure that all your tests are using postgresql_noproc fixture like that:

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

And run tests:

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

Basic database state for all tests

If you’ve got several tests that require common initialisation, you can to define a load and pass it to your custom postgresql process 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",
)

The way this will work is that the process fixture will populate template database, which in turn will be used automatically by client fixture to create a test database from scratch. Fast, clean and no dangling transactions, that could be accidentally rolled back.

Same approach will work with noproces fixture, while connecting to already running postgresql instance whether it’ll be on a docker machine or running remotely or locally.

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