Start docker container with PostgreSQL as a fixture in conjunction with pytest-postgresql plugin

Question

Recently I began to use pytest-postgresql plugin to test my app. I start docker container manually using docker run ... command, then I run my tests and it works like a charm. My conftest.py looks like this:

from pytest_postgresql import factories


postgresql_noproc = factories.postgresql_noproc(
    host="127.0.0.1",
    port=65432,
    password="supersecretpassword",
)
postgres_client = factories.postgresql("postgresql_noproc")

and here is a simple test:

from psycopg import sql
from psycopg.rows import dict_row


def test_tables_count(postgres_client):
    with postgres_client.cursor(row_factory=dict_row) as cursor:
        cursor.execute(
            sql.SQL(
                "SELECT COUNT(*) FROM information_schema.tables "
                "WHERE table_type = 'BASE TABLE'"
                " AND table_schema NOT IN ('pg_catalog', 'information_schema')"
            )
        )
        result = cursor.fetchone()
    assert result["count"] == 0

Now I want to start container automatically using docker-py library before initializing pytest-postgresql fixtures.

I added a fixture that would run a container:

import time

import docker
import pytest


@pytest.fixture
def psql_docker():
    client = docker.from_env()
    container = client.containers.run(
        image="postgres:12",
        auto_remove=True,
        environment=dict(
            POSTGRES_PASSWORD="supersecretpassword",
        ),
        name="test_postgres",
        ports={"5432/tcp": ("127.0.0.1", 65432)},
        detach=True,
        remove=True,
    )
    
    # Wait for the container to start
    # (actually I use more complex check to wait for container to start but it doesn't really matter)
    time.sleep(5)
    
    yield

    container.stop()

How can I make this fixture run before other factories? I understand that factories.postgresql_noproc and factories.postgresql fixture factories are module level entities and they initialize before psql_docker runs.

What I tried

Attempt #0

I tried to convert postgres_client to a custom fixture:

@pytest.fixture
def postgres_client(psql_docker):
    return factories.postgresql("postgresql_noproc")

but the test failed:

postgres_client = <function postgresql.<locals>.postgresql_factory at 0x7f810cb0d670>

    def test_tables_count(postgres_client):
>       with postgres_client.cursor(row_factory=dict_row) as cursor:
E       AttributeError: 'function' object has no attribute 'cursor'

tests/unit_postgres_so/test_psql_connection.py:6: AttributeError
=========================================================================================================================== short test summary info ============================================================================================================================
FAILED tests/unit_postgres_so/test_psql_connection.py::test_tables_count - AttributeError: 'function' object has no attribute 'cursor'

I guess this happens because postgres_client now returns a function object, not a fixture result.

Attempt #1

Almost the same, but in postgres_client fixture I return a function result:

@pytest.fixture
def postgres_client(psql_docker):
    return factories.postgresql("postgresql_noproc")()

Again, the test fails. This time with an error that says: "Fixture "postgresql_factory" called directly". I followed the link from that message, but I can't understand how it may help.

Of course, I can get rid of pytest-postgresql plugin and write my own fixture that'll be able to connect to dockerized PostgreSQL instance, but I really don't want to do it. Any help will be highly appreciated.

Answer 1

So under the hood pytest_postgresql creates an instance of a DatabaseJanitor for the noproc method. We can simply recreate this methodology and create our own fixture where we inject our dependency on docker in to it.

# conftest.py
import time

import docker
import psycopg
import pytest
from pytest_postgresql.janitor import DatabaseJanitor

POSTGRES_PASSWORD = "supersecretpassword"
PORT = 65432
USER = "postgres"
DB = "postgres"


@pytest.fixture(scope="session")
def psql_docker():
    client = docker.from_env()
    container = client.containers.run(
        image="postgres:12",
        auto_remove=True,
        environment=dict(
            POSTGRES_PASSWORD=POSTGRES_PASSWORD,
        ),
        name="test_postgres",
        ports={"5432/tcp": ("127.0.0.1", PORT)},
        detach=True,
        remove=True,
    )

    # Wait for the container to start
    # (actually I use more complex check to wait for container to start but it doesn't really matter)
    time.sleep(5)

    yield

    container.stop()


@pytest.fixture(scope="session")
def database(psql_docker):
    with DatabaseJanitor(
        user=USER,
        host="localhost",
        port=PORT,
        dbname=f"{DB}_tmpl",
        version=12,
        password=POSTGRES_PASSWORD,
    ):
        yield psycopg.connect(
            dbname=DB,
            user=USER,
            password=POSTGRES_PASSWORD,
            host="localhost",
            port=PORT,
        )

# test_file.py
from psycopg import sql
from psycopg.rows import dict_row


def test_tables_count(database):
    with database.cursor(row_factory=dict_row) as cursor:
        cursor.execute(
            sql.SQL(
                "SELECT COUNT(*) FROM information_schema.tables "
                "WHERE table_type = 'BASE TABLE'"
                " AND table_schema NOT IN ('pg_catalog', 'information_schema')"
            )
        )
        result = cursor.fetchone()
    assert result["count"] == 0

Now it works as expected.