Pytest Alembic

https://pytest-alembic.readthedocs.io/en/latest/
Admin
Pytest Alembic is a pytest plugin that provides built-in and custom tests for validating Alembic...
Tokens:18,561
Snippets:137
Trust Score:9.6
Update:1 month ago
Show doc for...
Tokens:
Raw
### Package structure example for all_models_register_on_metadata

Source: https://pytest-alembic.readthedocs.io/en/latest/experimental_tests.html

Example package structure showing models, other packages, and migrations directory. Used to illustrate the problem the test solves.

```text
package/
  models/
     __init__.py
     foo.py
     bar.py
     baz.py
  other_packages/
  other_modules.py

migrations/
   env.py

```

--------------------------------

### Install package and dev dependencies

Source: https://pytest-alembic.readthedocs.io/en/latest/contributing.html

Use this command to install the package along with its development dependencies.

```bash
# Installs the package along with its dev dependencies
make install
```

--------------------------------

### Build documentation

Source: https://pytest-alembic.readthedocs.io/en/latest/contributing.html

Commands to sync documentation dependencies and start a local sphinx-autobuild server.

```bash
uv sync --group docs
uv run sphinx-autobuild docs/source docs/build
```

--------------------------------

### Package structure example

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/experimental_tests.rst.txt

Example package structure showing models package and migrations directory, used to illustrate the problem the test solves.

```text
package/
  models/
     __init__.py
     foo.py
     bar.py
     baz.py
  other_packages/
  other_modules.py

migrations/
   env.py
```

--------------------------------

### Alembic Cookbook async env.py setup

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/asyncio.rst.txt

Standard async setup for Alembic's env.py, based on the Alembic Cookbook. It defines do_run_migrations, run_migrations_online, and the main entry point that runs migrations asynchronously. This is a prerequisite for running alembic commands with an async connection.

```python
import asyncio

# ... no change required to the rest of the code

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)

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


async def run_migrations_online():
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    connectable = AsyncEngine(
        engine_from_config(
            config.get_section(config.config_ini_section),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
            future=True,
        )
    )

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

    await connectable.dispose()


if context.is_offline_mode():
    run_migrations_offline()
else:
    asyncio.run(run_migrations_online())
```

--------------------------------

### Install pytest-alembic with pip

Source: https://pytest-alembic.readthedocs.io/en/latest/quickstart.html

Install the pytest-alembic plugin using pip.

```bash
pip install "pytest-alembic"

```

--------------------------------

### Migration history example

Source: https://pytest-alembic.readthedocs.io/en/latest/custom_data.html

Represents a sample migration history sequence used to illustrate the order of data insertion relative to upgrades.

```text
23256c7bf855 > fb4d3ab5f38d -> a5a9ccc4e535 -> b835ae9ff1d1
```

--------------------------------

### Install and run pytest-alembic with default tests

Source: https://pytest-alembic.readthedocs.io/en/latest/quickstart.html

Install the plugin with pip and run pytest with the --test-alembic flag to execute the built-in tests. The output shows the four default tests passing.

```bash
$ pip install pytest-alembic
$ pytest --test-alembic

...
::pytest_alembic/tests/model_definitions_match_ddl <- . PASSED           [ 25%]
::pytest_alembic/tests/single_head_revision <- . PASSED                  [ 50%]
::pytest_alembic/tests/up_down_consistency <- . PASSED                   [ 75%]
::pytest_alembic/tests/upgrade <- . PASSED                               [100%]

============================== 4 passed in 2.32s ===============================

```

--------------------------------

### Model definition in foo.py

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/experimental_tests.rst.txt

Example model definition in foo.py, showing a table with an integer primary key.

```python
from package.models import Base

class Foo(Base):
 __tablename__ = "foo"

 id = Column(types.Integer(), autoincrement=True, primary_key=True)
```

--------------------------------

### Shared alembic_engine and test engine

Source: https://pytest-alembic.readthedocs.io/en/latest/setup.html

Example showing how to share the same database engine between your tests and the alembic_engine fixture.

```python
import pytest
from pytest_mock_resources import create_postgres_fixture

pg = create_postgres_fixture()

@pytest.fixture
def alembic_engine(pg):
    return pg

def test_foo(pg, alembic_engine):
    assert pg is alembic_engine  # they're literally the same
    ...
```

--------------------------------

### Versatile env.py supporting sync and async migrations

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/asyncio.rst.txt

A reorganization of the Alembic Cookbook setup that allows both synchronous and asynchronous migration execution. It checks if the connection is an AsyncEngine and runs accordingly, fixing the test_downgrade_leaves_no_trace test. The outer call run_migrations_online() is included.

```python
from sqlalchemy.ext.asyncio.engine import AsyncEngine

def run_migrations_online():
    connectable = context.config.attributes.get("connection", None)

    if connectable is None:
        connectable = AsyncEngine(
            engine_from_config(
                context.config.get_section(context.config.config_ini_section),
                prefix="sqlalchemy.",
                poolclass=pool.NullPool,
                future=True,
            )
        )

    # Note, we decide whether to run asynchronously based on the kind of engine we're dealing with.
    if isinstance(connectable, AsyncEngine):
        asyncio.run(run_async_migrations(connectable))
    else:
        with connectable.connect() as connection:
            do_run_migrations(connection)


# Then use their setup for async connection/running of the migration
async def run_async_migrations(connectable):
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)

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


# But the outer layer still allows sychronous execution also.
run_migrations_online()
```

--------------------------------

### Independent alembic_engine and test engine

Source: https://pytest-alembic.readthedocs.io/en/latest/setup.html

Example showing how to ensure no interdependence between the alembic_engine and your own test engine using pytest-mock-resources.

```python
from pytest_mock_resources import create_postgres_fixture

pg = create_postgres_fixture()
alembic_engine = create_postgres_fixture()

def test_foo(pg, alembic_engine):  # two unique databases
    ...
```

--------------------------------

### Custom migration test with alembic_runner and alembic_engine

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/custom_tests.rst.txt

Use this snippet to write a custom test for a specific migration. It migrates up to just before the target migration, performs data setup, then migrates up one and verifies the data. The alembic_engine fixture provides a SQLAlchemy engine for direct queries.

```python
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,)]
```

--------------------------------

### env.py excerpt from alembic init

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/experimental_tests.rst.txt

Excerpt from env.py showing how target_metadata is set and configured, illustrating the problem of missing imports.

```python
...
from package.models import Base
target_metadata = Base.metadata
...
with connectable.connect() as connection:
    context.configure(connection=connection, target_metadata=target_metadata)
    ...
```

--------------------------------

### Enable all_models_register_on_metadata via setup.cfg/pytest.ini

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/experimental_tests.rst.txt

Configuration for setup.cfg or pytest.ini to enable the experimental test. Set pytest_alembic_include_experimental to all_models_register_on_metadata.

```ini
[pytest]
pytest_alembic_include_experimental = all_models_register_on_metadata
```

--------------------------------

### Models __init__.py with declarative_base

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/experimental_tests.rst.txt

Typical package __init__.py defining a declarative base and importing models. Note the try/except for sqlalchemy.orm import.

```python
import sqlalchemy
from sqlalchemy import Column, types

try:
    from sqlalchemy.orm import declarative_base
except ImportError:
    from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

from package.models import (
   foo,
   bar,
)
```

--------------------------------

### Models __init__.py with declarative_base

Source: https://pytest-alembic.readthedocs.io/en/latest/experimental_tests.html

Typical package __init__.py defining a declarative base and importing models. Shows how models are imported after base definition.

```python
import sqlalchemy
from sqlalchemy import Column, types

try:
    from sqlalchemy.orm import declarative_base
except ImportError:
    from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

from package.models import (
   foo,
   bar,
)

```

--------------------------------

### Run tests and linters

Source: https://pytest-alembic.readthedocs.io/en/latest/contributing.html

Commands to execute the CI-level test suite and run static analysis tools like ruff and mypy.

```bash
# Runs CI-level tests, with coverage reports
make test

# Runs ruff (lint and format check) and mypy
make lint
```

--------------------------------

### Enable downgrade_leaves_no_trace in setup.cfg/pytest.ini

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/experimental_tests.rst.txt

Adds the pytest_alembic_include_experimental option to the [pytest] section of setup.cfg or pytest.ini to enable the experimental test.

```ini
[pytest]
pytest_alembic_include_experimental = downgrade_leaves_no_trace
```

--------------------------------

### Create fixtures for multiple Alembic histories with create_alembic_fixture()

Source: https://pytest-alembic.readthedocs.io/en/latest/running.html

Use create_alembic_fixture() to define separate fixtures for each Alembic history, then write test functions that call the built-in tests with each fixture. The argument to create_alembic_fixture() is equivalent to alembic_config and may configure 'file', 'script_location', or other options. This approach avoids the --test-alembic flag and its automatic test insertion.

```python
from pytest_alembic import tests, create_alembic_fixture

# The argument here represents the equivalent to `alembic_config`. Depending
# on your setup, this may be configuring the "file" argument, "script_location",
# or some other way of configuring one or the other of your histories.
history_1 = create_alembic_fixture({"file": "alembic.ini"})

def test_single_head_revision_history_1(history_1):
    tests.test_single_head_revision(history_1)

def test_upgrade_history_1(history_1):
    tests.test_upgrade(history_1)

def test_model_definitions_match_ddl_history_1(history_1):
    tests.test_model_definitions_match_ddl(history_1)

def test_up_down_consistency_history_1(history_1):
    tests.test_up_down_consistency(history_1)

# The 2nd fixture, and the 2nd set of tests.
history_2 = create_alembic_fixture({"file": "history_2.ini"})

def test_single_head_revision_history_2(history_2):
    tests.test_single_head_revision(history_2)

def test_upgrade_history_2(history_2):
    tests.test_upgrade(history_2)

def test_model_definitions_match_ddl_history_2(history_2):
    tests.test_model_definitions_match_ddl(history_2)

def test_up_down_consistency_history_2(history_2):
    tests.test_up_down_consistency(history_2)

```

--------------------------------

### Using pytest-mock-resources for alembic_engine

Source: https://pytest-alembic.readthedocs.io/en/latest/setup.html

Recommended approach for managing ephemeral database instances using pytest-mock-resources.

```python
from pytest_mock_resources import create_postgres_fixture

alembic_engine = create_postgres_fixture()
```

--------------------------------

### Default context.configure block in env.py

Source: https://pytest-alembic.readthedocs.io/en/latest/setup.html

The configure block in env.py where additional options like compare_type, compare_server_default, and include_schemas can be added.

```python
with connectable.connect() as connection:
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        # This is where we want to add more options!
    )

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

```

--------------------------------

### Default env.py run_migrations_online

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/setup.rst.txt

The default implementation of run_migrations_online generated by Alembic.

```python
def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
```

--------------------------------

### Enable all_models_register_on_metadata via pyproject.toml

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/experimental_tests.rst.txt

Configuration for pyproject.toml to enable the experimental test. Set pytest_alembic_include_experimental to 'all_models_register_on_metadata'.

```toml
[tool.pytest.ini_options]
pytest_alembic_include_experimental = 'all_models_register_on_metadata'
```

--------------------------------

### Async engine fixtures for pytest-alembic

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/asyncio.rst.txt

Three ways to provide an async engine to the alembic_engine fixture: using create_async_engine, wrapping a sync engine with AsyncEngine, or using pytest-mock-resources with async_=True. The first two are plain fixtures that work without extra plugins; the pytest-mock-resources form requires an async-fixture plugin like pytest-asyncio.

```python
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine

@pytest.fixture
def alembic_engine(...):
    return create_async_engine(URL(...))


@pytest.fixture
def alembic_engine(...):
    engine = create_engine(URL(...))
    return AsyncEngine(engine)


# or, for example, with pytest-mock-resources
from pytest_mock_resources import create_postgres_fixture

alembic_engine = create_postgres_fixture(async_=True)
```

--------------------------------

### Manually run all_models_register_on_metadata test

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/experimental_tests.rst.txt

Manually import and execute the experimental test, providing the module/package directly to circumvent automatic detection.

```python
from pytest_alembic import tests

def test_all_models_register_on_metadata(alembic_runner):
   tests.experimental.test_all_models_register_on_metadata(alembic_runner, 'package.models')
```

--------------------------------

### Manually run downgrade_leaves_no_trace test

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/experimental_tests.rst.txt

Imports the experimental test and calls it with an alembic_runner fixture to execute the test directly, bypassing automatic detection.

```python
from pytest_alembic import tests

def test_downgrade_leaves_no_trace(alembic_runner):
   tests.experimental.test_downgrade_leaves_no_trace(alembic_runner)
```

--------------------------------

### Enable downgrade_leaves_no_trace in pyproject.toml

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/experimental_tests.rst.txt

Adds the pytest_alembic_include_experimental option to the [tool.pytest.ini_options] section of pyproject.toml to enable the experimental test.

```toml
[tool.pytest.ini_options]
pytest_alembic_include_experimental = 'downgrade_leaves_no_trace'
```

--------------------------------

### create_alembic_fixture

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/api.rst.txt

Function to create a custom alembic fixture.

```APIDOC
## create_alembic_fixture

### Description
Creates a custom alembic fixture for use in tests.

### Usage
Call to generate a fixture function.
```

--------------------------------

### Config class

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/api.rst.txt

Configuration class for pytest-alembic, with a from_raw_config method.

```APIDOC
## Config

### Description
Configuration class for pytest-alembic.

### Methods
- from_raw_config: Create a Config instance from a raw configuration.
```

--------------------------------

### Default run_migrations_online() from alembic env.py

Source: https://pytest-alembic.readthedocs.io/en/latest/setup.html

The default env.py snippet that alembic autogenerates. It creates a connectable using engine_from_config with the sqlalchemy prefix and NullPool.

```python
def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

```

--------------------------------

### Modified run_migrations_online() for pytest-alembic

Source: https://pytest-alembic.readthedocs.io/en/latest/setup.html

Modify the default snippet to use context.config.attributes.get("connection") so pytest-alembic can provide a connection at runtime. Falls back to engine_from_config if no connection is provided.

```python
def run_migrations_online():
    connectable = context.config.attributes.get("connection", None)

    if connectable is None:
        connectable = engine_from_config(
            context.config.get_section(context.config.config_ini_section),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
        )

```

--------------------------------

### Configure before_revision_data and at_revision_data in alembic_config fixture

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/custom_data.rst.txt

Defines a pytest fixture that returns an Alembic config with before_revision_data and at_revision_data keys. The before_revision_data inserts rows before upgrading to revision fb4d3ab5f38d, and at_revision_data inserts a row after upgrading to a5a9ccc4e535. Both options internally issue a commit during migration, so database state cannot be rolled back after migrations run.

```python
@pytest.fixture
   def alembic_config():
       return {
           "before_revision_data": {
               "fb4d3ab5f38d": [
                   {"__tablename__": "foo", "id": 1, "name": "foo"},
                   {"__tablename__": "bar", "id": 1, "name": "bar"},
               ],
           },
           "at_revision_data": {
               "a5a9ccc4e535": {
                   "__tablename__": "foo",
                   "id": 1,
                   "name": "foo2",
               },
           },
       }
```

--------------------------------

### Create separate alembic fixtures for multiple histories

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/running.rst.txt

Use create_alembic_fixture to define separate fixtures for each database history. Each fixture is configured with a different ini file, and then you define test functions that use the built-in tests from pytest_alembic.tests. This approach avoids the automatic test insertion from the --test-alembic flag.

```python
from pytest_alembic import tests, create_alembic_fixture

# The argument here represents the equivalent to `alembic_config`. Depending
# on your setup, this may be configuring the "file" argument, "script_location",
# or some other way of configuring one or the other of your histories.
history_1 = create_alembic_fixture({"file": "alembic.ini"})

def test_single_head_revision_history_1(history_1):
    tests.test_single_head_revision(history_1)

def test_upgrade_history_1(history_1):
    tests.test_upgrade(history_1)

def test_model_definitions_match_ddl_history_1(history_1):
    tests.test_model_definitions_match_ddl(history_1)

def test_up_down_consistency_history_1(history_1):
    tests.test_up_down_consistency(history_1)

# The 2nd fixture, and the 2nd set of tests.
history_2 = create_alembic_fixture({"file": "history_2.ini"})

def test_single_head_revision_history_2(history_2):
    tests.test_single_head_revision(history_2)

def test_upgrade_history_2(history_2):
    tests.test_upgrade(history_2)

def test_model_definitions_match_ddl_history_2(history_2):
    tests.test_model_definitions_match_ddl(history_2)

def test_up_down_consistency_history_2(history_2):
    tests.test_up_down_consistency(history_2)
```

--------------------------------

### Alembic env.py configure block

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/setup.rst.txt

The configuration block in env.py where additional autogeneration options can be added.

```python
with connectable.connect() as connection:
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        # This is where we want to add more options!
    )

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

--------------------------------

### Modified env.py run_migrations_online

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/setup.rst.txt

Modified run_migrations_online to allow pytest-alembic to provide a connection at runtime.

```python
def run_migrations_online():
    connectable = context.config.attributes.get("connection", None)

    if connectable is None:
        connectable = engine_from_config(
            context.config.get_section(context.config.config_ini_section),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
        )
```

--------------------------------

### Alternative import of built-in tests in tests/test_migrations.py

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/running.rst.txt

An alternative way to import test implementations directly from pytest_alembic to include them in your test suite without the --test-alembic flag.

```python
from pytest_alembic.tests import test_single_head_revision
from pytest_alembic.tests import test_upgrade
from pytest_alembic.tests import test_model_definitions_match_ddl
from pytest_alembic.tests import test_up_down_consistency
```

--------------------------------

### Default alembic_config implementation

Source: https://pytest-alembic.readthedocs.io/en/latest/setup.html

The default implementation of the alembic_config fixture, which can be overridden to configure the alembic context.

```python
from pytest_alembic.config import Config

@pytest.fixture
def alembic_config():
    """Override this fixture to configure the exact alembic context setup required.
    """
    return Config()
```

--------------------------------

### MigrationContext class

Source: https://pytest-alembic.readthedocs.io/en/latest/_sources/api.rst.txt

The object yielded by the alembic_runner fixture, providing methods to run migrations and check history.

```APIDOC
## MigrationContext

### Description
Provides methods to run migrations and inspect migration history.

### Methods
- run: Run migrations.
- check: Check migration state.
```

--------------------------------

### Configure custom data insertion with alembic_config

Source: https://pytest-alembic.readthedocs.io/en/latest/custom_data.html

Defines the alembic_config fixture to specify data to be inserted before or at specific migration revisions.

```python
@pytest.fixture
def alembic_config():
    return {
        "before_revision_data": {
            "fb4d3ab5f38d": [
                {"__tablename__": "foo", "id": 1, "name": "foo"},
                {"__tablename__": "bar", "id": 1, "name": "bar"},
            ],
        },
        "at_revision_data": {
            "a5a9ccc4e535": {
                "__tablename__": "foo",
                "id": 1,
                "name": "foo2",
            },
        },
    }
```

--------------------------------

### Default alembic_engine implementation

Source: https://pytest-alembic.readthedocs.io/en/latest/setup.html

The default implementation of the alembic_engine fixture, which uses an in-memory SQLite engine.

```python
@pytest.fixture
def alembic_engine():
    """Override this fixture to provide pytest-alembic powered tests with a database handle.
    """
    return sqlalchemy.create_engine("sqlite:///")
```
Tokens:
Raw