foxnouns/tests/conftest.py

import pytest
import pytest_asyncio
from sqlalchemy import delete, text

from foxnouns.db import Base
from foxnouns.settings import DATABASE

# Override the database name to the testing database
DATABASE["NAME"] = f"{DATABASE['NAME']}_test"


def pytest_collection_modifyitems(items):
    """Ensure that all async tests use the same event loop."""

    pytest_asyncio_tests = (
        item for item in items if pytest_asyncio.is_async_test(item)
    )
    session_scope_marker = pytest.mark.asyncio(scope="session")
    for async_test in pytest_asyncio_tests:
        async_test.add_marker(session_scope_marker, append=False)


@pytest.fixture(scope="session", autouse=True)
def setup():
    """Migrate the testing database to the latest migration, and once the tests complete, clear the database again."""

    from alembic import command, config
    from foxnouns.db.sync import engine

    cfg = config.Config("alembic.ini")
    cfg.attributes["connection"] = engine.connect()
    command.upgrade(cfg, "head")

    yield

    with engine.begin() as session:
        Base.metadata.drop_all(session)
        session.execute(text("DROP TABLE alembic_version"))
        session.commit()


@pytest.fixture(scope="function", autouse=True)
def clean_tables_after_tests():
    """Clean tables after every test."""

    yield

    from foxnouns.db.sync import engine

    with engine.begin() as session:
        for table in reversed(Base.metadata.sorted_tables):
            session.execute(delete(table))
        session.commit()


@pytest_asyncio.fixture(scope="session", autouse=True)
async def app():
    from foxnouns.app import app

    return app