unwind/tests/conftest.py

import asyncio

import pytest
import pytest_asyncio

from unwind import db


@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.run_until_complete(loop.shutdown_asyncgens())
    loop.run_until_complete(loop.shutdown_default_executor())
    loop.close()


@pytest_asyncio.fixture(scope="session")
async def shared_conn():
    """A database connection, ready to use."""
    await db.open_connection_pool()

    async with db.new_connection() as c:
        db._test_connection = c
        yield c
        db._test_connection = None

    await db.close_connection_pool()


@pytest_asyncio.fixture
async def conn(shared_conn: db.Connection):
    """A transacted database connection, will be rolled back after use."""
    async with db.transacted(shared_conn, force_rollback=True):
        yield shared_conn