API

Fixtures

alembic_runner

pytest_alembic.plugin.fixtures.alembic_runner(alembic_config, alembic_engine)

Produce the primary alembic migration context in which to execute alembic tests.

This fixture allows authoring custom tests which are specific to your particular migration history.

Examples

>>> def test_specific_migration(alembic_runner):
...     alembic_runner.migrate_up_to('xxxxxxx')
...     assert ...

alembic_config

pytest_alembic.plugin.fixtures.alembic_config()

Override this fixture to configure the exact alembic context setup required.

The return value of this fixture can be one of a few types. :rtype: Union[Dict[str, Any], Config, Config]

• If you’re only configuring alembic-native configuration, a alembic.config.Config object is accepted as configuration. This largely leaves pytest-alembic out of the setup, so depending on your settings, might be the way to go.

• If you only have a couple of options to set, you might choose to return a Dict.

  The following common alembic config options are accepted as keys.

  • file/config_file_name (commonly alembic.ini)

  • script_location

  • sqlalchemy.url

  • target_metadata

  • process_revision_directives

  • include_schemas

  Additionally you can send a file key (akin to alembic -c), should your alembic.ini be otherwise named.

  Note that values here, represent net-additive options on top of what you might already have configured in your env.py. You should generally prefer to configure your env.py however you like it and omit such options here.

  You may also use this dict to set pytest-alembic specific features:

  • before_revision_data

  • at_revision_data

  • minimum_downgrade_revision

• You can also directly return a Config class instance. This is your only option if you want to use both pytest-alembic specific features and construct your own alembic.config.Config.

Examples

>>> @pytest.fixture
... def alembic_config():
...     return {'file': 'migrations.ini'}

>>> @pytest.fixture
... def alembic_config():
...     alembic_config = alembic.config.Config()
...     alembic_config.set_main_option("script_location", ...)
...     return alembic_config

Config

class pytest_alembic.config.Config(config_options=<factory>, alembic_config=None, before_revision_data=None, at_revision_data=None, minimum_downgrade_revision=None, skip_revisions=None)

Pytest-alembic configuration options.

• config_options: Meant to simplify the creation of alembic.config.Config

  objects. Supply keys common to customization in alembic configuration. For example:

  • file/config_file_name (commonly alembic.ini)

  • script_location

  • sqlalchemy.url

  • target_metadata

  • process_revision_directives

  • include_schemas

• Both before_revision_data and at_revision_data are described in detail in Custom data.

• minimum_downgrade_revision can be used to set a lower bound on the downgrade migrations which are run built-in tests like test_up_down_consistency and test_downgrade_leaves_no_trace.

• skip_revisions can be used to avoid executing specific revisions, particularly if they are slow and you can guarantee to yourself that the difference in the resulting migrations history wont have a meaningful effect. Note that skipping migrations can be “dangerous”, because either DDL or data differences could lead to migrations which pass in tests, but fail in practice.

For example:

>>> import pytest

>>> @pytest.fixture
... def alembic_config():
...    return Config(minimum_downgrade_revision='abcde12345')

This would essentially short-circuit and avoid running the downgrade migrations including and below this migration.

Note

If a downgrade raises a NotImplementedError, it will have the same effect as a minimum_downgrade_revision, but will emit a warning suggesting the use of this feature instead.

classmethod from_raw_config(raw_config=None)

Adapt between pre-produced alembic config and raw config options.

Allows one to specify raw pytest-alembic config options through raw dictionary, as well as being flexible enough to allow a literal alembic Config object.

Examples

>>> Config.from_raw_config()
Config(config_options={}, alembic_config=None, before_revision_data=None, at_revision_data=None, minimum_downgrade_revision=None, skip_revisions=None)

>>> Config.from_raw_config({'minimum_downgrade_revision': 'abc123'})
Config(config_options={}, alembic_config=None, before_revision_data=None, at_revision_data=None, minimum_downgrade_revision='abc123', skip_revisions=None)

>>> Config.from_raw_config(Config(minimum_downgrade_revision='abc123'))
Config(config_options={}, alembic_config=None, before_revision_data=None, at_revision_data=None, minimum_downgrade_revision='abc123', skip_revisions=None)

alembic_engine

pytest_alembic.plugin.fixtures.alembic_engine()

Override this fixture to provide pytest-alembic powered tests with a database handle.

create_alembic_fixture

pytest_alembic.plugin.fixtures.create_alembic_fixture(raw_config=None)

Create a new fixture alembic_runner-like fixture.

In many cases, this function should not be strictly necessary. You can generally rely solely on the --test-alembic flag, automatic insertion of tests, and the alembic_runner() fixture.

However this may be useful in some situations:

• If you would generally prefer to avoid the --test-alembic flag and automatic test insertion, this is the function for you!

• If you have multiple alembic histories and therefore require more than one fixture, you will minimally need to use this for the 2nd history (if not both)

Examples

>>> from pytest_alembic import tests
>>>
>>> alembic = create_alembic_fixture()
>>>
>>> def test_upgrade_head(alembic):
...     tests.test_upgrade_head(alembic)
>>>
>>> def test_specific_migration(alembic):
...     alembic_runner.migrate_up_to('xxxxxxx')
...     assert ...

Config can also be supplied similarly to the alembic_config() fixture.

>>> alembic = create_alembic_fixture({'file': 'migrations.ini'})

Alembic Runner

The object yielded into a test from an alembic_runner fixture is the MigrationContext

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.

class pytest_alembic.history.AlembicHistory(map, revisions, revision_indices, revisions_by_index)

classmethod parse(revision_map)

Extract the set of migration revision hashes from alembic’s notion of the history.

Return type:

AlembicHistory

class pytest_alembic.revision_data.RevisionData(before_revision_data, at_revision_data)

Describe the data which should exist at given revisions when performing upgrades.

classmethod from_config(config)

Produce a RevisionData from raw configuration from alembic_config().

get_at(revision)

Yield the individual data insertions which should occur upon reaching the given revision.

Return type:

Union[Dict, List[Dict]]

get_before(revision)

Yield the individual data insertions which should occur before the given revision.

Return type:

Union[Dict, List[Dict]]

class pytest_alembic.revision_data.RevisionSpec(data)

Describe a set of valid database data at a set of revisions.

get(revision)

Get the database data described at a particular revision.

Return type:

Union[Dict, List[Dict]]

classmethod parse(data)

Parse a raw dict structure into a RevisionSpec.