Start Docker container when running Pytest

Consider starting a Docker container before running a Pytest test. It makes it easier to perform integration tests with peripheral middleware, and has the advantage of obtaining environment, test data, and idempotency.

As an example, start the PostgreSQL container at the time of test execution and try to input data.

Remarks

Before the code example, let's consider the restrictions of using the Docker container ** in the unit test.

--The Docker daemon needs to be running
-> Naturally, but important. One more hurdle for cases where you want to test run in a container (?) --It seems that you can do it by building a container that can do Docker in Docker (unverified). --Reference: Docker in Docker Better Practice

--You need to wait until the container starts --The container itself starts up quickly, but it often takes a few seconds for the internal process to complete and ** become available. --As the number of test functions increases, the execution time of all test cases will increase significantly. --It is better to use a container image that is as lightweight as possible. -** Watch the standard output of the container and devise to wait properly until it is ready **

Library used

--pytest: Unit test --docker: Docker API wrapper --docker --PyPI

• SQLAlchemy: ORM --psycopg2: PostgreSQL connection driver

$ pip install docker pytest SQLAlchemy psycopg2-binary

Constitution

├── main.py
├── models.py
└── tests
    ├── __init__.py
    ├── conftest.py
    └── test_main.py

Main processing / DB model

• models.py

models.py

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base


Base = declarative_base()


class Staff(Base):
    __tablename__ = 'staff'
    id = Column(Integer, primary_key=True)
    name = Column(String)

• main.py

main.py

from sqlalchemy.orm import Session
from models import Staff


def add_staff(engine, name):
    session = Session(bind=engine)
    staff = Staff()
    staff.name = name
    session.add(staff)
    session.commit()

Test code

The point is to make the engine of SQLAlchemy a fixture so that it can be used in test functions.

• conftest.py

tests/conftest.py

import time
import pytest
import docker
from sqlalchemy import create_engine


@pytest.fixture()
def pg_conf():
    """Manage PostgreSQL settings"""
    host = '127.0.0.1'
    port = 5432
    dbname = 'pytest'
    user = 'testuser'
    password = 'test'
    pg_conf = {'host': host,
               'port': port,
               'dbname': dbname,
               'user': user,
               'password': password,
               'url': f'postgresql://{user}:{password}@{host}/{dbname}'}
    return pg_conf


@pytest.fixture()
def engine(pg_conf):
    return create_engine(pg_conf['url'])


@pytest.fixture(autouse=True)
def pg_container(pg_conf):
    """Start the PostgreSQL container"""
    client = docker.from_env()
    container = client.containers.run(image='postgres:11.6-alpine',
                                      tty=True,
                                      detach=True,
                                      auto_remove=True,
                                      environment={'POSTGRES_DB': pg_conf['dbname'],
                                                   'POSTGRES_USER': pg_conf['user'],
                                                   'POSTGRES_PASSWORD': pg_conf['password']},
                                      ports={pg_conf['port']: '5432'})
    #Wait until the container is ready
    while True:
        log = container.logs(tail=1)
        if 'database system is ready to accept connections' in log.decode():
            break
        time.sleep(0.5)
    yield  #Transition to test here
    container.kill()

I'm checking the standard output in the container, but an error occurs if the wait interval is too short (0.4 seconds or less). It seems better to have a waiting time with a little grace.

• test_main.py

test_main.py

from sqlalchemy.orm import Session
from models import Base, Staff
from main import add_staff


def test_add(engine):
    #Add 1 record
    Base.metadata.create_all(bind=engine)  #Create table
    add_staff(engine=engine,
              name='alice')
    #Check the added record
    session = Session(bind=engine)
    assert session.query(Staff.id).filter_by(name='alice').first() == (1,)
    session.close()

Execution result

$ pytest --setup-show tests/ -v -s
========================================= test session starts =========================================platform linux -- Python 3.8.1, pytest-5.3.3, py-1.8.1, pluggy-0.13.1 -- /home/skokado/.local/share/virtualenvs/sandbox-pTebjwBw/bin/python3.8
cachedir: .pytest_cache
rootdir: ***
collected 1 item

tests/test_pg.py::test_add
        SETUP    F pg_conf
        SETUP    F pg_container (fixtures used: pg_conf)
        SETUP    F engine (fixtures used: pg_conf)
        tests/test_main.py::test_add (fixtures used: engine, pg_conf, pg_container)PASSED
        TEARDOWN F engine
        TEARDOWN F pg_container
        TEARDOWN F pg_conf

========================================== 1 passed in 2.00s ==========================================