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
    alembic_runner.migrate_up_before('xyz123')

    # 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.

    alembic_runner.migrate_up_one()

    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

str

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

List[str]

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

Insert data into a given table.

Parameters
  • 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.

managed_upgrade(dest_revision)

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

migrate_down_before(revision)

Migrate down to, but not including the given revision.

migrate_down_one()

Migrate down by exactly one revision.

migrate_down_to(revision)

Migrate down to, and including the given revision.

migrate_up_before(revision)

Migrate up to, but not including the given revision.

migrate_up_one()

Migrate up by exactly one revision.

migrate_up_to(revision)

Migrate up to, and including the given revision.

raw_command(*args, **kwargs)

Execute a raw alembic command.

roundtrip_next_revision()

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.

Parameters
  • 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.