Несмотря на большую популярность библиотеки 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.
