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, history, config)

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

property current: str

Get the list of revision heads.

generate_revision(process_revision_directives=None, prevent_file_generation=True, autogenerate=False, **kwargs)

Generate a test revision.

If prevent_file_generation is True, 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 on disk.

property heads: List[str]

Get the list of revision heads.

Result is cached for the lifetime of the MigrationContext.

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_downgrade(dest_revision)

Perform an downgrade, one migration at a time.

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.

refresh_history()

Refresh the context’s version of the alembic history.

Note this is not done automatically to avoid the expensive reevaluation step which can make long histories take seconds longer to evaluate for each test.

Return type:

AlembicHistory

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.