Issues with async tests

[einarjohnson]

Hello, I am trying to write unit tests using the pytest-alembic package where I migrate my db schema into an in-memory sqlite db.

I have defined fixtures like this:

from pytest_alembic.config import Config
from sqlalchemy.ext.asyncio import create_async_engine

@pytest.fixture
def alembic_config():
    return Config()

@pytest.fixture
def alembic_engine():
    return create_async_engine("sqlite+aiosqlite:///")

My env.py file is like this:

import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config

from alembic import context
from myproject.database.models import Base

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = Base.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.

def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()

def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations() -> None:
    """In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    connectable = context.config.attributes.get("connection", None)
    if connectable is None:
        connectable = async_engine_from_config(
            config.get_section(config.config_ini_section, {}),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
        )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()

def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""

    asyncio.run(run_async_migrations())

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

When I run the following test:

import pytest
from sqlalchemy.ext.asyncio.engine import AsyncEngine

@pytest.mark.asyncio
async def test_alembic_migration(alembic_runner, alembic_engine: AsyncEngine):
    alembic_runner.migrate_up_to("heads")

I get this error:

alembic_runner = MigrationContext(command_executor=CommandExecutor(alembic_config=<alembic.config.Config object at 0x28d0b9190>, stdout...c_config=None, before_revision_data=None, at_revision_data=None, minimum_downgrade_revision=None, skip_revisions=None))
alembic_engine = <sqlalchemy.ext.asyncio.engine.AsyncEngine object at 0x28ae4cfc0>

    @pytest.mark.asyncio
    async def test_alembic_migration(alembic_runner, alembic_engine: AsyncEngine):
>       alembic_runner.migrate_up_to("heads")

tests/pytest_alembic/test_database.py:7:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.venv/lib/python3.11/site-packages/pytest_alembic/runner.py:196: in migrate_up_to
    return self.managed_upgrade(revision, return_current=return_current)
.venv/lib/python3.11/site-packages/pytest_alembic/runner.py:146: in managed_upgrade
    current = self.current
.venv/lib/python3.11/site-packages/pytest_alembic/runner.py:82: in current
    self.command_executor.execute_fn(get_current)
.venv/lib/python3.11/site-packages/pytest_alembic/executor.py:41: in execute_fn
    self.script.run_env()
.venv/lib/python3.11/site-packages/alembic/script/base.py:585: in run_env
    util.load_python_file(self.dir, "env.py")
.venv/lib/python3.11/site-packages/alembic/util/pyfiles.py:93: in load_python_file
    module = load_module_py(module_id, path)
.venv/lib/python3.11/site-packages/alembic/util/pyfiles.py:109: in load_module_py
    spec.loader.exec_module(module)  # type: ignore
<frozen importlib._bootstrap_external>:940: in exec_module
    ???
<frozen importlib._bootstrap>:241: in _call_with_frames_removed
    ???
alembic/env.py:91: in <module>
    run_migrations_online()
alembic/env.py:85: in run_migrations_online
    asyncio.run(run_async_migrations())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

main = <coroutine object run_async_migrations at 0x105baeb60>

    def run(main, *, debug=None):
        """Execute the coroutine and return the result.

        This function runs the passed coroutine, taking care of
        managing the asyncio event loop and finalizing asynchronous
        generators.

        This function cannot be called when another asyncio event loop is
        running in the same thread.

Any help/directions would be greatly appreciated, not entirely sure what I am doing wrong here but it obviously looks like there is something wrong in how I am setting the asyncio framework up.

[DanCardin]

You will need to adjust your env.py slightly, so that the plugin can inject the connection, documented here (or you would either need to supply a Config that set the engine details appropriately)

[einarjohnson]

@DanCardin , ah of course. Thank you kindly for such a quick response, I think I have everything sorted out on my end now.

[DanCardin]

nice! closing then. feel free to comment again if something comes up

[himself65]

I'm getting RuntimeError: asyncio.run() cannot be called from a running event loop how to fix it?🤔