60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
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
|