Тестируем слой БД в Python с использованием pytest и testcontainers
Автор: Z55Несмотря на большую популярность библиотеки 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 и реализуем следующие базы:
Модули приложения
Дабы не перегружать пост кодом, уберу модули приложения под спойлер.
Модули приложения
Тестовые БД
Очевидно, что создаваемая тестовая БД должна отражать структуру продуктовой БД, по крайней мере она должна содержать тестируемые объекты. Для создания тестовых БД, будем использовать 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 вполне может быть выходом в данной ситуации.
Скачать данный проект можно с моего репозитория