Function init_tests

tests/lifespan.py:90–133

(config, scope="module")

def init_tests(config, scope="module"):
    @pytest_asyncio.fixture(autouse=True, scope=scope)
    async def create_database():

        # Drop and create tables in a single connection to avoid event loop issues
        async with config.engine.begin() as conn:

            def setup_tables(connection):  # pragma: no cover
                validate_cross_schema_constraints(
                    config.metadata, connection.dialect.name
                )
                drop_tables(connection, config)
                drop_schemas(connection, config)
                create_schemas(connection, config)
                config.metadata.create_all(connection)

            await conn.run_sync(setup_tables)

        # For PostgreSQL and MySQL, recreate engine to avoid event loop conflicts
        # asyncpg and aiomysql are strict about event loops
        if config.engine.dialect.name in ("postgresql", "mysql"):  # pragma: no cover
            await config.engine.dispose()
            config._original_engine = config.engine
            config.engine = create_async_engine(ASYNC_DATABASE_URL)

        yield

        # Restore the original engine if it was swapped
        if hasattr(config, "_original_engine"):  # pragma: no cover
            await config.engine.dispose()
            config.engine = config._original_engine
            delattr(config, "_original_engine")

        async with config.engine.begin() as conn:

            def teardown_tables(connection):  # pragma: no cover
                drop_tables(connection, config)
                drop_schemas(connection, config)

            await conn.run_sync(teardown_tables)

        await config.engine.dispose()

    return create_database