55 lines
2.0 KiB
Python
55 lines
2.0 KiB
Python
"""Помощники интеграционных тестов с PostgreSQL: временная БД внутри того же сервера, что и `OPENBAO__DATABASE`.
|
||
|
||
Фикстура `scratch_db_url` создаёт пустую БД `armcore_scratch_<hex>`, отдаёт её SQLAlchemy-URL (драйвер asyncpg) и
|
||
в конце удаляет. Рабочую БД (`armka`) тесты не трогают. Подключается плагином в `conftest.py`.
|
||
"""
|
||
|
||
import uuid
|
||
from collections.abc import AsyncIterator
|
||
|
||
import asyncpg
|
||
import pytest_asyncio
|
||
from sqlalchemy.engine import URL, make_url
|
||
|
||
from src.core.config import settings
|
||
|
||
|
||
def base_url() -> URL:
|
||
"""SQLAlchemy-URL рабочей БД из настроек (`.env.test`)."""
|
||
return make_url(settings.OPENBAO.DATABASE.DATABASE_URL)
|
||
|
||
|
||
def _asyncpg_dsn(url: URL) -> str:
|
||
return url.set(drivername="postgresql").render_as_string(hide_password=False)
|
||
|
||
|
||
async def create_scratch_db(name: str) -> URL:
|
||
"""Создать пустую БД `name` на сервере рабочей БД; вернуть SQLAlchemy-URL на неё."""
|
||
url = base_url()
|
||
conn = await asyncpg.connect(_asyncpg_dsn(url))
|
||
try:
|
||
await conn.execute(f'CREATE DATABASE "{name}"')
|
||
finally:
|
||
await conn.close()
|
||
return url.set(database=name)
|
||
|
||
|
||
async def drop_scratch_db(name: str) -> None:
|
||
"""Удалить БД `name` (закрывая оставшиеся подключения)."""
|
||
conn = await asyncpg.connect(_asyncpg_dsn(base_url()))
|
||
try:
|
||
await conn.execute(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)')
|
||
finally:
|
||
await conn.close()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def scratch_db_url() -> AsyncIterator[str]:
|
||
"""URL пустой временной БД (asyncpg); БД удаляется после теста."""
|
||
name = f"armcore_scratch_{uuid.uuid4().hex[:12]}"
|
||
url = await create_scratch_db(name)
|
||
try:
|
||
yield url.render_as_string(hide_password=False)
|
||
finally:
|
||
await drop_scratch_db(name)
|