pythonpostgresqldockerpytestpsycopg3

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


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.


Solution

  • 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.