39 lines
1.2 KiB
Python
39 lines
1.2 KiB
Python
"""Shared pytest fixtures.
|
|
|
|
The auth integration tests hit the *real* Postgres database (per
|
|
``DATABASE_URL``), so they persist data between runs. Without cleanup a
|
|
second run of the suite fails with 409 conflicts because the users it
|
|
registered on the first run already exist. The session-scoped fixture below
|
|
truncates all application tables once before the first test, making the suite
|
|
re-runnable.
|
|
|
|
Truncation order doesn't strictly matter because ``CASCADE`` follows foreign
|
|
keys, but the tables are listed explicitly for clarity.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from app.config import settings
|
|
|
|
TEST_TABLES = ("assets", "worlds", "sessions", "refresh_tokens", "users")
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def _reset_database():
|
|
"""Truncate all application tables once per test session."""
|
|
|
|
async def _truncate() -> None:
|
|
engine = create_async_engine(settings.DATABASE_URL)
|
|
async with engine.begin() as conn:
|
|
await conn.execute(
|
|
text(f"TRUNCATE TABLE {', '.join(TEST_TABLES)} CASCADE")
|
|
)
|
|
await engine.dispose()
|
|
|
|
asyncio.run(_truncate())
|
|
yield
|