conftest.py

import pytest
import psycopg2
from pytest_docker_containers.fixtures import docker_container_fixture


pytest_plugins = [
    'pytest_docker_containers.plugin',
]


def pg_check_callback(container):
    """
    A health checker for the PostgreSQL docker container.

    :param container:
        A docker container with running PostgreSQL instance to check
        the health state. If no exceptions supplied, the container
        is considered as "live." Otherwise, increase the timeout a
        little and repeat the health check.

    :type container: docker.
    """

    network = container.attrs['HostConfig']
    with psycopg2.connect(
        host='127.0.0.1',  # TODO how I do take that hardcode off?
        port=network['PortBindings']['5432/tcp'][0]['HostPort'],
        user='postgres',
        database='postgres'
    ) as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT 1')
            assert cur.fetchone() == (1,)

postgresql_container = docker_container_fixture(
    images=['postgres:9.6.5-alpine'],
    port='5432',
    health_check_callback=pg_check_callback,
)


@pytest.fixture(scope='session')
def postgresql_dsn(postgresql_container):
    host_cfg = postgresql_container.attrs['HostConfig']

    return 'postgres://{user}@{host}:{port}/{db_name}'.format(
        user='postgres',
        host='127.0.0.1',
        port=host_cfg['PortBindings']['5432/tcp'][0]['HostPort'],
        db_name='postgres',
    )


@pytest.fixture(scope='session')
def pg_client(postgresql_dsn):
    "Returns an opened connection to the running postgresql container"
    with psycopg2.connect(postgresql_dsn) as conn:
        yield conn

pytest_docker_containers__fixtures.py

import time
import typing
import pytest


def docker_container_fixture(
    image: typing.Optional[str]=None,
    images: typing.List[str]=None,
    port: typing.Optional[str]=None,
    ports: typing.List[int]=None,
    health_check_callback: typing.Optional[typing.Callable]=None,
    container_setup_callback: typing.Optional[typing.Callable]=None,
    container_teardown_callback: typing.Optional[typing.Callable]=None,
    environment: typing.Optional[dict]=None
):
    """
    Returns a container fixture function
    """

    if all(i is None for i in (image, images)):
        raise TypeError(
            'Either `image` (a single docker image name) or `images` (a '
            'list of ones) is to be filled'
        )

    if all(p is None for p in (port, ports)):
        raise TypeError(
            'Either `port` (a single TCP port inside a container) '
            'or `ports` (a list of ones) is to be filled'
        )

    def format(container):
        return container

    def setup(container):
        if not callable(container_setup_callback):
            return
        container_setup_callback(container)

    def teardown(container):
        if not callable(container_teardown_callback):
            return
        container_teardown_callback(container)
        
    def wait_until_ready(container):
        if not callable(health_check_callback):
            return
        timeout = 0.001
        for i in range(100):
            try:
                health_check_callback(container)
                break
            except Exception:
                time.sleep(timeout)
                timeout *= 1.125
        else:
            pytest.skip(f'Unable create a docker container '
                        f'during {timeout}s')

    fixture_kwargs = {}

    if images:
        fixture_kwargs.update(
            params=images,
            ids=[f'{img.split(":")[0].title()} version {img.split(":")[1]}'
                 for img in images]
        )

    if not ports:
        ports = [port]

    @pytest.fixture(scope='session', **fixture_kwargs)
    def fixture_body(request, session_id, unused_port, docker_client):
        container = docker_client.containers.run(
            image=getattr(request, 'param', None) or image,
            name=f'{image}-{session_id}'.replace(':', '-').replace('/', '-'),
            ports={f'{port}/tcp': unused_port() for port in ports},
            remove=True,
            detach=True,
            environment=environment or {},
        )
        wait_until_ready(container)
        setup(container)
        yield format(container)
        teardown(container)
        container.remove(force=True)
    return fixture_body

pytest_docker_containers__plugin.py

import socket
import uuid
import typing
import docker
import pytest


def pytest_report_header(config):
    return 'docker containers plugin enabled'


def pytest_addoption(parser):
    docker_group = parser.getgroup(name='docker',
                                   description=(
                                       'Temporary Docker containers plugin'
                                   ))
    docker_group.addoption('--docker-base-url',
                           required=False,
                           metavar='URL',
                           action='store',
                           dest='docker_base_url',
                           default=None,
                           help='An URL to docker connect to.')


    @pytest.fixture(scope='session')
def session_id() -> str:
    """The session-wide identifier to avoid container name clashes."""
    return f'{uuid.uuid4()}'


@pytest.fixture(scope='session')
def unused_port() -> typing.Callable[[], int]:
    """Returns a random unused TCP port number"""
    def inner() -> int:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.bind(('', 0))
            return sock.getsockname()[1]
    return inner


@pytest.fixture(scope='session')
def docker_client(request):
    """A high-level Docker client instance"""
    base_url = request.config.getoption('docker_base_url')
    return docker.DockerClient(base_url=base_url)

    

test_db_connection.py

@pytest.fixture
def pg(pg_client):
    "A high-level PostgreSQL client instance"
    return postgresql.Pg(pg_client)


@pytest.mark.parametrize('key', ['host', 'port'])
def test_create_ok(pg, key):
    res = pg.create_resource()
    dsn = pg.conn.get_dsn_parameters()

    assert getattr(res, key) == dsn[key]


def test_cannot_create_two_users_with_same_names(pg):
    pg.create_user(user='foo', password='bar')
    with pytest.raises(postgresql.CannotCreateUser):
        pg.create_user(user='foo', password='bar')


def test_cannot_create_two_databases_with_same_names(pg):
    pg.create_user('john', password='123')
    pg.create_database('db_name', user='john')

    with pytest.raises(postgresql.CannotCreateDatabase):
        pg.create_database('db_name', user='john')