Writing Custom Tests

Honestly, there’s not much to it by this point!

from sqlalchemy import text

def test_gnarly_migration_xyz123(alembic_runner, alembic_engine):
    # Migrate up to, but not including this new migration

    # Perform some very specific data setup, because this migration is sooooo complex.
    # ...
    alembic_runner.insert_into('tablename', dict(id=1, name='foo'))
    # Or you can optionally accept the `alembic_engine` fixture, which is a
    # sqlalchemy engine object, with which you can do whatever setup you'd like.


    with alembic_engine.connect() as conn:
        rows = conn.execute(text("SELECT id from foo")).fetchall()

    assert rows == [(1,)]

alembic_runner has all sorts of convenience methods for altering the state of the database for your test:

class pytest_alembic.runner.MigrationContext(command_executor, revision_data, connection_executor, config, history, connection=None)

Within a given environment/execution context, executes alembic commands.

property current: str

Get the list of revision heads.

Return type


generate_revision(process_revision_directives=None, **kwargs)

Generate a test revision.

The final act of this process raises a RevisionSuccess, which is used as a sentinal to indicate the revision was generated successfully, while not actually finishing the generation of the revision file.

property heads: List[str]

Get the list of revision heads.

Result is cached for the lifetime of the MigrationContext.

Return type


insert_into(table, data=None, revision=None)

Insert data into a given table.

  • table (Optional[str]) – The name of the table to insert data into

  • data (Union[Dict, List, None]) – The data to insert. This is eventually passed through to SQLAlchemy’s Table class values method, and so should accept either a list of dict`s representing a list of rows, or a `dict representing one row.

  • revision – The revision of MetaData to use as the table definition for the insert.


Perform an upgrade, one migration at a time, inserting static data at the given points.


Migrate down to, but not including the given revision.


Migrate down by exactly one revision.


Migrate down to, and including the given revision.


Migrate up to, but not including the given revision.


Migrate up by exactly one revision.


Migrate up to, and including the given revision.

raw_command(*args, **kwargs)

Execute a raw alembic command.


Upgrade, downgrade then upgrade.

This is meant to ensure that the given revision is idempotent.

table_at_revision(name, *, revision=None, schema=None)

Return a reference to a sqlalchemy.Table at the given revision.

  • name – The name of the table to produce a sqlalchemy.Table for.

  • revision – The revision of the table to return.

  • schema – The schema of the table.