45 lines
1.2 KiB
Python
45 lines
1.2 KiB
Python
import os
|
|
import sys
|
|
from pathlib import Path
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
|
|
|
|
ROOT_DIR = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT_DIR / "src"))
|
|
|
|
from logger import log
|
|
|
|
|
|
def pytest_sessionstart():
|
|
os.environ['MYSQL_HOST'] = 'localhost'
|
|
os.environ['MYSQL_USER'] = 'root'
|
|
os.environ['MYSQL_PASSWORD'] = 'root'
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
async def db():
|
|
# from database.transaction import get_session
|
|
#
|
|
# async with get_session() as db:
|
|
# yield db
|
|
# await db.rollback()
|
|
|
|
|
|
import database
|
|
from database.config import create_db_engine
|
|
|
|
test_engine = create_db_engine()
|
|
test_async_session = async_sessionmaker(test_engine, expire_on_commit=True, class_=AsyncSession)
|
|
|
|
database.config.engine = test_engine
|
|
database.async_session = test_async_session
|
|
|
|
async with test_async_session() as session:
|
|
async with session.begin():
|
|
try:
|
|
yield session
|
|
await session.rollback()
|
|
except Exception as e:
|
|
log.exception("Error", exc_info=e)
|
|
await session.rollback()
|