100 lines
2.6 KiB
Python
100 lines
2.6 KiB
Python
import logging
|
|
import os
|
|
|
|
import xdist
|
|
from pytest import Session, fixture, hookimpl
|
|
from sqlalchemy.ext.asyncio import (
|
|
AsyncEngine,
|
|
AsyncSession,
|
|
async_sessionmaker,
|
|
create_async_engine,
|
|
)
|
|
|
|
from fastfood_two.application.config import Config
|
|
from tests.utils import db_creator, db_deleter, get_test_settings
|
|
|
|
|
|
def pytest_configure(config):
|
|
worker_id = os.environ.get("PYTEST_XDIST_WORKER")
|
|
if worker_id is not None:
|
|
logging.basicConfig(
|
|
format=config.getini("log_file_format"),
|
|
filename=f"tests_{worker_id}.log",
|
|
level=config.getini("log_file_level"),
|
|
)
|
|
|
|
|
|
@hookimpl(trylast=True)
|
|
def pytest_sessionstart(session: Session) -> None:
|
|
"""Hook запуска приложения в отдельном процессе."""
|
|
worker_id = xdist.get_xdist_worker_id(session)
|
|
|
|
import asyncio
|
|
|
|
db_name = f"fastfood_two_test_db_{worker_id}"
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
loop.run_until_complete(db_creator(name=db_name))
|
|
loop.close()
|
|
|
|
|
|
@hookimpl(trylast=True)
|
|
def pytest_sessionfinish(session: Session) -> None:
|
|
"""Hook остановки приложения."""
|
|
worker_id = xdist.get_xdist_worker_id(session)
|
|
import asyncio
|
|
|
|
db_name = f"fastfood_two_test_db_{worker_id}"
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
loop.run_until_complete(db_deleter(name=db_name))
|
|
loop.close()
|
|
|
|
|
|
@fixture(scope="session")
|
|
def app_settings() -> Config:
|
|
worker_id = os.environ.get("PYTEST_XDIST_WORKER")
|
|
if worker_id == None:
|
|
worker_id = "master"
|
|
settings = get_test_settings(db_name=f"fastfood_two_test_db_{worker_id}")
|
|
return settings
|
|
|
|
|
|
@fixture
|
|
def get_loop():
|
|
import asyncio
|
|
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@fixture
|
|
def test_engine(app_settings):
|
|
return create_async_engine(url=app_settings.db.url)
|
|
|
|
|
|
@fixture
|
|
def test_sessionmaker(test_engine):
|
|
return async_sessionmaker(test_engine, expire_on_commit=False)
|
|
|
|
|
|
@fixture
|
|
async def test_session(test_sessionmaker):
|
|
|
|
async with test_sessionmaker() as session:
|
|
try:
|
|
yield session
|
|
finally:
|
|
pass
|
|
|
|
|
|
@fixture
|
|
def app(app_settings, test_engine, test_sessionmaker, test_session):
|
|
from fastfood_two.presentation.fastapi_backend.main import app_factory
|
|
|
|
app = app_factory()
|
|
app.dependency_overrides[Config] = lambda: app_settings
|
|
app.dependency_overrides[AsyncEngine] = lambda: test_engine
|
|
app.dependency_overrides[async_sessionmaker[AsyncSession]] = lambda: test_sessionmaker
|
|
app.dependency_overrides[AsyncSession] = lambda: test_session
|
|
return app
|