Z55 18 июл 2022 в 21:01

Тестируем слой БД в Python с использованием pytest и testcontainers

15 мин

15K

Технотекст 2022

Несмотря на большую популярность библиотеки testcontainers в мире java, информации в сети по её применению в python практически нет. Даная статья - попытка ликвидировать этот пробел. Я не буду подробно рассказывать про pytest и testcontainers, что это такое можно почитать в интернете, я просто покажу пример того, как можно собрать это воедино.

В качестве БД будем использовать PostgreSQL. В сети есть следующий пример использования testcontainers с PostgreSQL:

with PostgresContainer("postgres:9.5") as postgres:
  e = sqlalchemy.create_engine(postgres.get_connection_url())
  result = e.execute("select version()")

Да, не много. Поэтому давайте разовьём этот пример до применения в реальном приложении.

Структура проекта

В качестве примера, создадим часть приложения, которое будет оперировать данными в двух БД и протестируем его. В нашем примере мы реализуем три метода бизнес-логики, которые будут работать с БД, напишем классы и методы для взаимодействия БД, напишем сами тесты и всё то, что потребуется для подготовки и запуска тестовой среды. Структура проекта выглядит следующим образом:

Небольшие пояснения к структуре:

db_services - пакет, в котором располагаются базовые процедуры для работы с БД
processing - пакет, содержащий процедуры реализующие бизнес-логику работы приложения
tests - пакет, содержащий всё необходимо для создания контейнера тестового SQL - сервера и наполнения его данными, а также сами тесты.

За основу баз данных возьмём кусочек от демонстрационной базы данных от PostgresPro и реализуем следующие базы:

Модули приложения

Дабы не перегружать пост кодом, уберу модули приложения под спойлер.

Модули приложения

engine_factory.py
В модуле реализована фабрика соединений ко всем БД приложения, по паттерну singleton. Такой подход позволяет использовать одни и те же соединения к БД из любых модулей приложения, без необходимости выполнения затратных операций в виде создания новых соединений.

import sqlalchemy.engine
from sqlalchemy import create_engine


class MetaSingleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(MetaSingleton, cls).__call__(*args, **kwargs)
        return cls._instances[cls]


class EngineFactory(metaclass=MetaSingleton):
    connections, db_urls = ({},) * 2
    user, passw, stand, db_name = (None,) * 4

    def get_engine(self, db_name, schema_name=None) -> sqlalchemy.engine.Engine:
        self.db_name = db_name

        if None in (self.user, self.stand, self.db_name):
            raise ValueError('Не заданы обязательные параметры: stand, user, db_name')

        if self.connections.get((db_name, schema_name)):
            return self.connections.get((db_name, schema_name))
        else:
            url = self.get_postgres_url(db_name)
            if schema_name:
                self.connections[(db_name, schema_name)] = create_engine(url, echo=False, echo_pool=False,
                                                                         connect_args={
                                                                             'options': f'-csearch_path={schema_name}'})
            else:
                self.connections[(db_name, schema_name)] = create_engine(url, echo=False, echo_pool=False)
        return self.connections[(db_name, schema_name)]

    def dispose_engines(self) -> None:
        for engine in self.connections.values():
            engine.dispose()
        self.connections = {}

    def add_db(self, base_name, url):
        self.db_urls[base_name] = url

    def get_postgres_url(self, base_name) -> str:
        stand = self.stand.lower()

        if stand == 'localhost':
            return self.db_urls.get(base_name) if self.db_urls.get(base_name) else ValueError(
                f'''URL для параметров stand={stand}, db_name='{base_name}' не найден ''')

db_service.py
Здесь напишем два основных метода взаимодействия с БД (GET и POST), которые упростят обращение к БД через стандартные SQL-запросы из других модулей.

from .engine_factory import EngineFactory
engine = EngineFactory()


def get_from_postgres(sql, db_name, schema_name=None) -> list:
    result = []
    pg_engine = engine.get_engine(db_name=db_name, schema_name=schema_name)
    try:
        with pg_engine.connect() as connection:
            cursor = connection.execution_options(stream_result=True).execute(sql)
            for row in cursor:
                result.append(list(row))
    except Exception as e:
        raise RuntimeError(f'Ошибка при обращении к БД: {e}')
    return result


def post_to_postgres(sql, db_name, schema_name=None) -> int:
    pg_engine = engine.get_engine(db_name=db_name, schema_name=schema_name)
    rows_processed = 0
    try:
        with pg_engine.connect() as connection:
            cursor = connection.execution_options(stream_result=True, isolation_level='AUTOCOMMIT').execute(sql)
            rows_processed = cursor.rowcount
            cursor.close()
    except Exception as e:
        raise RuntimeError(f'Ошибка при выполнении операции {sql} в БД: {e}')
    return rows_processed

airline.py
В данном модуле опишем методы, реализующие бизнес-логику в БД Airline

from db_services.db_service import get_from_postgres
from db_services.db_service import post_to_postgres

DB_NAME = 'airline'

# Установить статуса рейса
def set_flight_status(flight_id, status) -> int:
    sql = '''
    update airline.flights
        set status = '%s'
        where flight_id = %d
    ''' % (status, flight_id)

    try:
        return post_to_postgres(sql=sql, db_name=DB_NAME)
    except Exception as e:
        raise RuntimeError(e)

# Получить статус рейса
def get_flight_status(flight_id) -> list:
    sql = '''
    select status
        from airline.flights
        where flight_id = %d
    ''' % flight_id

    try:
        return get_from_postgres(sql=sql, db_name=DB_NAME)
    except Exception as e:
        raise RuntimeError(e)

bookings.py
В данном модуле опишем методы, реализующие бизнес-логику в БД Bookings

from db_services.db_service import get_from_postgres

DB_NAME = 'bookings'

# Получить список пассажиров, траты которых более limit
def get_premium_psg_list(limit) -> list:
    sql = '''
            select passenger_name, sum(amount)
                    from bookings.tickets
                             join bookings.ticket_flights using (ticket_id)
                    group by 1
                    having sum(amount) > %d ''' % limit

    try:
        return get_from_postgres(sql=sql, db_name=DB_NAME)
    except Exception as e:
        raise RuntimeError(e)

Тестовые БД

Очевидно, что создаваемая тестовая БД должна отражать структуру продуктовой БД, по крайней мере она должна содержать тестируемые объекты. Для создания тестовых БД, будем использовать ORM с декларативным определением классов.

Структура и тестовые данные для БД Airline

airline_db.py

from sqlalchemy import Column, String, INTEGER, TEXT, ForeignKey
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class Aircrafts(Base):
    __tablename__ = 'aircrafts'
    __table_args__ = {'schema': 'airline'}

    aircraft_code = Column(String(3), nullable=False, primary_key=True, comment='Код самолета, IATA')
    model = Column(TEXT, nullable=False, comment='Модель самолета')
    range = Column(INTEGER, nullable=False, comment='Максимальная дальность полета, км')


class Flights(Base):
    __tablename__ = 'flights'
    __table_args__ = {'schema': 'airline'}

    flight_id = Column(INTEGER, nullable=False, primary_key=True, comment='Идентификатор рейса')
    flight_no = Column(String(10), nullable=False, comment='Номер рейса')
    aircraft_code = Column(String(3), ForeignKey('airline.aircrafts.aircraft_code'), nullable=False,
                           comment='Код самолета, IATA')
    status = Column(String(20), nullable=False, comment='Статус рейса')


AIRCRAFTS_ROWS = [
    {
        "aircraft_code": "773",
        "model": "Boeing 777-300",
        "range": 11100
    },
    {
        "aircraft_code": "763",
        "model": "Boeing 767-300",
        "range": 7900
    },
    {
        "aircraft_code": "SU9",
        "model": "Sukhoi Superjet-100",
        "range": 3000
    },
    {
        "aircraft_code": "320",
        "model": "Airbus A320-200",
        "range": 5700
    },
    {
        "aircraft_code": "321",
        "model": "Airbus A321-200",
        "range": 5600
    },
    {
        "aircraft_code": "319",
        "model": "Airbus A319-100",
        "range": 6700
    },
    {
        "aircraft_code": "733",
        "model": "Boeing 737-300",
        "range": 4200
    },
    {
        "aircraft_code": "CN1",
        "model": "Cessna 208 Caravan",
        "range": 1200
    },
    {
        "aircraft_code": "CR2",
        "model": "Bombardier CRJ-200",
        "range": 2700
    }
]

FLIGHTS_ROWS = [
    {
        "flight_id": 32959,
        "flight_no": "PG0550",
        "aircraft_code": "CR2",
        "status": "On Time"
    },
    {
        "flight_id": 28948,
        "flight_no": "PG0242",
        "aircraft_code": "SU9",
        "status": "Arrived"
    },
    {
        "flight_id": 33116,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "On Time"
    },
    {
        "flight_id": 33117,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Arrived"
    },
    {
        "flight_id": 33111,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Scheduled"
    },
    {
        "flight_id": 28929,
        "flight_no": "PG0242",
        "aircraft_code": "SU9",
        "status": "Arrived"
    },
    {
        "flight_id": 33052,
        "flight_no": "PG0359",
        "aircraft_code": "CR2",
        "status": "Cancelled"
    },
    {
        "flight_id": 33043,
        "flight_no": "PG0359",
        "aircraft_code": "CR2",
        "status": "On Time"
    },
    {
        "flight_id": 33118,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Arrived"
    },
    {
        "flight_id": 30007,
        "flight_no": "PG0386",
        "aircraft_code": "SU9",
        "status": "Delayed"
    },
    {
        "flight_id": 28913,
        "flight_no": "PG0242",
        "aircraft_code": "SU9",
        "status": "Arrived"
    },
    {
        "flight_id": 33099,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Cancelled"
    },
    {
        "flight_id": 32207,
        "flight_no": "PG0425",
        "aircraft_code": "CN1",
        "status": "Departed"
    },
    {
        "flight_id": 33115,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Arrived"
    },
    {
        "flight_id": 33107,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Scheduled"
    },
    {
        "flight_id": 32806,
        "flight_no": "PG0080",
        "aircraft_code": "CN1",
        "status": "Cancelled"
    },
    {
        "flight_id": 32961,
        "flight_no": "PG0550",
        "aircraft_code": "CR2",
        "status": "Cancelled"
    },
    {
        "flight_id": 31611,
        "flight_no": "PG0494",
        "aircraft_code": "CN1",
        "status": "Delayed"
    },
    {
        "flight_id": 28895,
        "flight_no": "PG0242",
        "aircraft_code": "SU9",
        "status": "Arrived"
    },
    {
        "flight_id": 30961,
        "flight_no": "PG0004",
        "aircraft_code": "CR2",
        "status": "Delayed"
    },
    {
        "flight_id": 31946,
        "flight_no": "PG0193",
        "aircraft_code": "CN1",
        "status": "Departed"
    },
    {
        "flight_id": 28904,
        "flight_no": "PG0242",
        "aircraft_code": "SU9",
        "status": "Arrived"
    },
    {
        "flight_id": 28915,
        "flight_no": "PG0242",
        "aircraft_code": "SU9",
        "status": "Arrived"
    },
    {
        "flight_id": 33114,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Arrived"
    },
    {
        "flight_id": 33119,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Scheduled"
    },
    {
        "flight_id": 32863,
        "flight_no": "PG0080",
        "aircraft_code": "CN1",
        "status": "On Time"
    },
    {
        "flight_id": 33112,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Scheduled"
    },
    {
        "flight_id": 32898,
        "flight_no": "PG0147",
        "aircraft_code": "SU9",
        "status": "On Time"
    },
    {
        "flight_id": 28939,
        "flight_no": "PG0242",
        "aircraft_code": "SU9",
        "status": "Arrived"
    },
    {
        "flight_id": 33121,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Scheduled"
    },
    {
        "flight_id": 31363,
        "flight_no": "PG0619",
        "aircraft_code": "CN1",
        "status": "Delayed"
    },
    {
        "flight_id": 32083,
        "flight_no": "PG0708",
        "aircraft_code": "733",
        "status": "Departed"
    },
    {
        "flight_id": 28935,
        "flight_no": "PG0242",
        "aircraft_code": "SU9",
        "status": "Arrived"
    },
    {
        "flight_id": 28942,
        "flight_no": "PG0242",
        "aircraft_code": "SU9",
        "status": "Arrived"
    },
    {
        "flight_id": 31867,
        "flight_no": "PG0304",
        "aircraft_code": "SU9",
        "status": "Departed"
    },
    {
        "flight_id": 28912,
        "flight_no": "PG0242",
        "aircraft_code": "SU9",
        "status": "Arrived"
    },
    {
        "flight_id": 32871,
        "flight_no": "PG0616",
        "aircraft_code": "SU9",
        "status": "Cancelled"
    },
    {
        "flight_id": 32937,
        "flight_no": "PG0147",
        "aircraft_code": "SU9",
        "status": "Departed"
    },
    {
        "flight_id": 33120,
        "flight_no": "PG0063",
        "aircraft_code": "CR2",
        "status": "Arrived"
    },
    {
        "flight_id": 32247,
        "flight_no": "PG0604",
        "aircraft_code": "CR2",
        "status": "Delayed"
    }
]

AIRLINE_ROWS = {
    Aircrafts: AIRCRAFTS_ROWS,
    Flights: FLIGHTS_ROWS
}

Структура и тестовые данные для БД Bookings

bookings_db.py

from sqlalchemy import Column, String, INTEGER, BIGINT, TEXT, ForeignKey, PrimaryKeyConstraint, NUMERIC
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class Ticket(Base):
    __tablename__ = 'tickets'
    __table_args__ = {'schema': 'bookings'}

    ticket_id = Column(BIGINT, nullable=False, unique=True, autoincrement=True, primary_key=True,
                       comment='Номер билета')
    passenger_id = Column(String(20), nullable=False, comment='Идентификатор пассажира')
    passenger_name = Column(TEXT, nullable=False, comment='Имя пассажира')


class Ticket_Flights(Base):
    __tablename__ = 'ticket_flights'
    __table_args__ = {'schema': 'bookings'}

    ticket_id = Column(BIGINT, ForeignKey('bookings.tickets.ticket_id'), nullable=False, unique=True,
                       comment='Номер билета')
    flight_id = Column(INTEGER, nullable=False, comment='Идентификатор рейса')
    amount = Column(NUMERIC(10, 2), nullable=False, comment='Стоимость перелета')
    PrimaryKeyConstraint(ticket_id, flight_id)


class Boarding_Passes(Base):
    __tablename__ = 'boarding_passes'
    __table_args__ = {'schema': 'bookings'}

    boarding_no = Column(BIGINT, nullable=False, unique=True, autoincrement=True, primary_key=True,
                         comment='Номер посадочного талона')
    ticket_id = Column(BIGINT, ForeignKey('bookings.tickets.ticket_id'), nullable=False, unique=True,
                       comment='Номер билета')
    seat_no = Column(String(4), nullable=False, comment='Номер места')


TICKET_ROWS = [
    {
        "ticket_id": 5432000987,
        "passenger_id": "8149 604011",
        "passenger_name": "VALERIY TIKHONOV"
    },
    {
        "ticket_id": 5432000988,
        "passenger_id": "8499 420203",
        "passenger_name": "EVGENIYA ALEKSEEVA"
    },
    {
        "ticket_id": 5432000989,
        "passenger_id": "1011 752484",
        "passenger_name": "ARTUR GERASIMOV"
    },
    {
        "ticket_id": 5432000990,
        "passenger_id": "4849 400049",
        "passenger_name": "ALINA VOLKOVA"
    },
    {
        "ticket_id": 5432000991,
        "passenger_id": "6615 976589",
        "passenger_name": "MAKSIM ZHUKOV"
    },
    {
        "ticket_id": 5432000992,
        "passenger_id": "2021 652719",
        "passenger_name": "NIKOLAY EGOROV"
    },
    {
        "ticket_id": 5432000993,
        "passenger_id": "0817 363231",
        "passenger_name": "TATYANA KUZNECOVA"
    },
    {
        "ticket_id": 5432000994,
        "passenger_id": "2883 989356",
        "passenger_name": "IRINA ANTONOVA"
    },
    {
        "ticket_id": 5432000995,
        "passenger_id": "3097 995546",
        "passenger_name": "VALENTINA KUZNECOVA"
    },
    {
        "ticket_id": 5432000996,
        "passenger_id": "6866 920231",
        "passenger_name": "POLINA ZHURAVLEVA"
    },
    {
        "ticket_id": 5432000997,
        "passenger_id": "6030 369450",
        "passenger_name": "ALEKSANDR TIKHONOV"
    },
    {
        "ticket_id": 5432000998,
        "passenger_id": "8675 588663",
        "passenger_name": "ILYA POPOV"
    },
    {
        "ticket_id": 5432000999,
        "passenger_id": "0764 728785",
        "passenger_name": "ALEKSANDR KUZNECOV"
    },
    {
        "ticket_id": 5432001000,
        "passenger_id": "8954 972101",
        "passenger_name": "VSEVOLOD BORISOV"
    },
    {
        "ticket_id": 5432001001,
        "passenger_id": "6772 748756",
        "passenger_name": "NATALYA ROMANOVA"
    },
    {
        "ticket_id": 5432001002,
        "passenger_id": "7364 216524",
        "passenger_name": "ANTON BONDARENKO"
    },
    {
        "ticket_id": 5432001003,
        "passenger_id": "3635 182357",
        "passenger_name": "VALENTINA NIKITINA"
    },
    {
        "ticket_id": 5432001004,
        "passenger_id": "8252 507584",
        "passenger_name": "ALLA TARASOVA"
    },
    {
        "ticket_id": 5432001005,
        "passenger_id": "1026 982766",
        "passenger_name": "OKSANA MOROZOVA"
    },
    {
        "ticket_id": 5432001006,
        "passenger_id": "7107 950192",
        "passenger_name": "GENNADIY GERASIMOV"
    },
    {
        "ticket_id": 5432001007,
        "passenger_id": "4765 014996",
        "passenger_name": "RAISA KONOVALOVA"
    }
]

TICKET_FLIGHTS_ROWS = [
    {
        "ticket_id": 5432000987,
        "flight_id": 28935,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432000988,
        "flight_id": 28935,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432000990,
        "flight_id": 28939,
        "amount": 18500.00
    },
    {
        "ticket_id": 5432000989,
        "flight_id": 28939,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432000991,
        "flight_id": 28913,
        "amount": 18500.00
    },
    {
        "ticket_id": 5432000992,
        "flight_id": 28913,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432000993,
        "flight_id": 28913,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432000994,
        "flight_id": 28912,
        "amount": 6800.00
    },
    {
        "ticket_id": 5432000995,
        "flight_id": 28912,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432000996,
        "flight_id": 28929,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432000998,
        "flight_id": 28904,
        "amount": 18500.00
    },
    {
        "ticket_id": 5432000999,
        "flight_id": 28904,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432000997,
        "flight_id": 28904,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432001001,
        "flight_id": 28895,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432001000,
        "flight_id": 28895,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432001002,
        "flight_id": 28895,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432001003,
        "flight_id": 28948,
        "amount": 18500.00
    },
    {
        "ticket_id": 5432001004,
        "flight_id": 28948,
        "amount": 6800.00
    },
    {
        "ticket_id": 5432001005,
        "flight_id": 28942,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432001007,
        "flight_id": 28915,
        "amount": 6200.00
    },
    {
        "ticket_id": 5432001006,
        "flight_id": 28915,
        "amount": 6200.00
    }
]

BOARDING_PASSES_ROWS = [
    {
        "boarding_no": 5432000959,
        "ticket_id": 5432000997,
        "seat_no": "19F"
    },
    {
        "boarding_no": 5432000962,
        "ticket_id": 5432000989,
        "seat_no": "18E"
    },
    {
        "boarding_no": 5432000963,
        "ticket_id": 5432001005,
        "seat_no": "17C"
    },
    {
        "boarding_no": 5432000965,
        "ticket_id": 5432001006,
        "seat_no": "16C"
    },
    {
        "boarding_no": 5432000969,
        "ticket_id": 5432000995,
        "seat_no": "17A"
    },
    {
        "boarding_no": 5432000970,
        "ticket_id": 5432000993,
        "seat_no": "19E"
    },
    {
        "boarding_no": 5432000974,
        "ticket_id": 5432000988,
        "seat_no": "10E"
    },
    {
        "boarding_no": 5432000977,
        "ticket_id": 5432000987,
        "seat_no": "7A"
    },
    {
        "boarding_no": 5432000978,
        "ticket_id": 5432001002,
        "seat_no": "12C"
    },
    {
        "boarding_no": 5432000979,
        "ticket_id": 5432001000,
        "seat_no": "11D"
    },
    {
        "boarding_no": 5432000981,
        "ticket_id": 5432001001,
        "seat_no": "11A"
    },
    {
        "boarding_no": 5432000982,
        "ticket_id": 5432001007,
        "seat_no": "8D"
    },
    {
        "boarding_no": 5432000983,
        "ticket_id": 5432000999,
        "seat_no": "8F"
    },
    {
        "boarding_no": 5432000984,
        "ticket_id": 5432000996,
        "seat_no": "14A"
    },
    {
        "boarding_no": 5432000986,
        "ticket_id": 5432000994,
        "seat_no": "6F"
    },
    {
        "boarding_no": 5432000987,
        "ticket_id": 5432000992,
        "seat_no": "5F"
    },
    {
        "boarding_no": 5432000988,
        "ticket_id": 5432000990,
        "seat_no": "3F"
    },
    {
        "boarding_no": 5432000989,
        "ticket_id": 5432001004,
        "seat_no": "6F"
    },
    {
        "boarding_no": 5432000990,
        "ticket_id": 5432000991,
        "seat_no": "1D"
    },
    {
        "boarding_no": 5432000996,
        "ticket_id": 5432000998,
        "seat_no": "2C"
    },
    {
        "boarding_no": 5432001000,
        "ticket_id": 5432001003,
        "seat_no": "2C"
    }
]

BOOKINGS_ROWS = {
    Ticket: TICKET_ROWS,
    Ticket_Flights: TICKET_FLIGHTS_ROWS,
    Boarding_Passes: BOARDING_PASSES_ROWS
}

Следующим шагом создадим класс, который будет собственно поднимать из нужного нам Docker - образа контейнер, запускать в нём сервер баз данных, создавать сами базы данных и наполнять их тестовыми данными, которые мы ранее определили в модулях airline_db.py и bookings_db.py

db_test.py

from sqlalchemy import create_engine
from sqlalchemy.dialects.postgresql import insert
from testcontainers.postgres import PostgresContainer

from db_services.engine_factory import EngineFactory
from .airline_db import Base as Airline_Base, AIRLINE_ROWS
from .bookings_db import Base as Bookings_Base, BOOKINGS_ROWS


class MetaSingleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(MetaSingleton, cls).__call__(*args, **kwargs)
        return cls._instances[cls]


class TestBases(metaclass=MetaSingleton):
    db = None
    main_url = None

    def __init__(self, db_image_name):
        __engine = EngineFactory()
        __engine.stand = 'localhost'
        # Создание контейнера из образа DB_IMAGE
        __postgres_container = PostgresContainer(image=db_image_name)
        self.db = __postgres_container.start()
        self.main_url = self.db.get_connection_url()

        __BASES = {'airline': {'class': Airline_Base, 'rows': AIRLINE_ROWS},
                   'bookings': {'class': Bookings_Base, 'rows': BOOKINGS_ROWS}
                   }

        # Создание баз, схем, наполнение данными
        for __base_name, __base_data in __BASES.items():
            self.create_base(base_name=__base_name)
            __engine.user, __engine.passw = 'test', 'test'
            __url = __engine.get_postgres_url(base_name=__base_name)
            self.create_schema(schema_name=__base_name, url=__url)
            __db_engine = __engine.get_engine(__base_name)
            __base_data.get('class').metadata.create_all(__db_engine)

            for __cls, __rows in __base_data.get('rows').items():
                __db_engine.execute(insert(__cls).values(__rows))

    def create_base(self, base_name):
        __engine = create_engine(self.main_url)
        __connection = __engine.connect()
        __connection.execution_options(isolation_level='AUTOCOMMIT').execute(f'create database {base_name}')
        __host, __port = self.main_url.replace('postgresql+psycopg2://test:test@', '').replace('/test', '').split(':')
        __new_base_url = f'postgresql+psycopg2://test:test@{__host}:{__port}/{base_name}'
        #Добавляем соединение с новой базой в EngineFactory
        __engine = EngineFactory()
        __engine.add_db(base_name=base_name, url=__new_base_url)

    def create_schema(self, url, schema_name):
        __engine = create_engine(url)
        __connection = __engine.connect()
        __connection.execution_options(isolation_level='AUTOCOMMIT').execute(f'create schema {schema_name}')

Здесь также реализуем singleton, т.к. мы хотим чтобы у нас поднимался только один testcontainers. Осталось дописать сами тесты:

tests.py

import pytest

from processing.airline import set_flight_status, get_flight_status
from processing.bookings import get_premium_psg_list
from .db_test import TestBases


@pytest.fixture(scope="session", autouse=True)
def test_db():
    # Этот блок будет выполнен перед запуском тестов
    test_base = TestBases(db_image_name='postgres:11.8')
    yield
    # Этот блок будет выполнен после окончания работы тестов
    test_base.db.stop()


# Тест метода processing.bookings.get_premium_psg_list()
# В текущих тестовых данных, для limit=10000, корректный результат == 4
def test_get_premium_psg_list(test_db):
    assert len(get_premium_psg_list(limit=10000)) == 4


# Тест метода processing.airline.get_flight_status()
# Для flight_id=33043 корректный результат 'On Time'
def test_get_flight_status_before(test_db):
    assert get_flight_status(flight_id=33043) == [['On Time']]


# Тест метода processing.airline.set_flight_status()
# В таблице airline.flights только одна запись с flight_id=33043, поэтому корректный результат - 1
# !!!Тест меняет состояние тестовой среды!!!
def test_set_flight_status(test_db):
    assert set_flight_status(flight_id=33043, status='Delayed') == 1


# Тест метода processing.airline.get_flight_status()
# После выполнения теста test_set_flight_status(test_db) состояние тестовой среды изменилось.
# Корректный результат теста для flight_id=33043 - 'Delayed'
def test_get_flight_status_after(test_db):
    assert get_flight_status(flight_id=33043) == [['Delayed']]


# тест, для случая если нужно оставить активным докер-контейнер после завершения работы тестов
# def test_debug(test_db):
#     while True:
#         pass

Здесь мы определили 4 теста, на которых и будем выполнять тестирование. Но что более важно, здесь мы определили фикстуру test_db(), внутри которой выполняется подготовка тестовой среды. Тестовую среду pytest будет создавать перед каждым тестом, который её использует, но т.к. мы указали scope="session", то подготовка тестовой среды будет производиться один раз для всей сессии выполнения тестов. И если какой-либо из тестов будет изменять состояние БД, то следующий тест будет использовать данные изменённой тестовой среды. Это необходимо учитывать. В частности, этот принцип используется в наших примерах.

Запускаем тесты и радуемся зелёными галочками в Test Result :)

После выполнения всех тестов, testcontainers завершит работу созданного контейнера и удалит созданные данные. Бывает полезно "придержать" тестовую БД на некоторое время, чтобы можно было залезть в БД из обычной IDE, чтобы выполнить пару-тройку SQL-запросов. Для этого нужно просто раскомментировать тест test_debug(test_db), который выполняясь в бесконечном цикле, позволит получить доступ к локальной БД под логином test и паролем test. Порт можно подсмотреть в Docker Desktop

либо из консоли:

# Список запущенных контейнеров:
docker ps                                                              
CONTAINER ID   IMAGE           COMMAND                  CREATED          STATUS          PORTS                     NAMES
46fb9a865f58   postgres:11.8   "docker-entrypoint.s…"   13 minutes ago   Up 13 minutes   0.0.0.0:56517->5432/tcp   clever_einstein

 # Получаем порты нужного контейнера 
docker port 46fb9a865f58                                               
5432/tcp -> 0.0.0.0:56517

Итоги

Мы только что создали проект, в котором протестировали БД слой приложения, с помощью testcontainers и pytest. Конечно, если у вас есть возможность тестирования на реальной БД или на её реплике, то смысл использования testcontainers теряется, а подготовка тестовых баз и тестовых данных становится ненужной тратой рабочего времени. Альтернативой testcontainers также может стать создание отдельного сервера БД с нужными объектами. Но если ничего такого под рукой нет, а тестирование необходимо выполнять, testcontainers вполне может быть выходом в данной ситуации.

Скачать данный проект можно с моего репозитория GitHub.

Теги:

Хабы:

Python

Тестируем слой БД в Python с использованием pytest и testcontainers

Структура проекта

Модули приложения

Тестовые БД

Структура и тестовые данные для БД Airline

Структура и тестовые данные для БД Bookings

Итоги

Публикации

Истории

Работа

Ближайшие события