Skip to content

api.md

Latest commit

 

History

History
359 lines (250 loc) · 8.79 KB

api.md

File metadata and controls

359 lines (250 loc) · 8.79 KB

API Reference

Complete reference for all public classes, methods, and CLI commands.

MRTConfig

from pytest_mrt import MRTConfig

Configuration object passed to pytest_configure to set up migration rollback testing.

Parameters

Parameter Type Default Description
alembic_ini str "alembic.ini" Path to your alembic.ini file
db_url str "" SQLAlchemy database URL for the test database
seed_rows int 3 Number of rows to seed per table during rollback verification
skip dict[str, str] {} Revisions to skip, with documented reasons
severity_overrides dict[str, str] {} Override severity of specific risk patterns
custom_seeds dict[str, Callable] {} Custom seed functions per table
custom_checks list[Callable] [] Additional static analysis check functions
migration_timeout int | None None Per-migration timeout in seconds (None = no limit)

Example

# conftest.py
import os
from pytest_mrt import MRTConfig

def pytest_configure(config):
    config._mrt_config = MRTConfig(
        alembic_ini="alembic.ini",
        db_url=os.environ["TEST_DATABASE_URL"],
        seed_rows=5,
        skip={
            "abc123": "Intentional one-way data migration. Reviewed 2025-01-15. See ADR-007."
        },
        severity_overrides={
            "INDEX without CONCURRENTLY": "error",
        },
        custom_seeds={
            "users": lambda: [{"id": 1, "name": "Alice", "email": "alice@example.com"}],
        },
    )

skip

Documents why a specific revision is exempt from rollback testing. Skipped revisions appear in reports as "skipped" (not failed).

skip={
    "1a2b3c4d": "RunPython data migration — irreversible by design. ADR-12.",
    "5e6f7a8b": "Adds NOT NULL column to 500M-row table — zero-downtime handled externally.",
}

severity_overrides

Promotes warnings to errors (or demotes errors to warnings) for specific risk pattern names. Pattern names match the pattern field in the JSON output of mrt check.

severity_overrides={
    "INDEX without CONCURRENTLY": "error",      # treat as error in your org
    "noop downgrade": "warning",                 # already handled by your deploy process
}

custom_checks

Each function receives a MigrationAST and returns a list of RiskWarning objects. Custom checks run in addition to the built-in checks.

from pytest_mrt.core.ast_analyzer import MigrationAST
from pytest_mrt.core.detector import RiskWarning

def check_no_truncate(m: MigrationAST) -> list[RiskWarning]:
    if "TRUNCATE" in m.source.upper():
        return [RiskWarning(
            revision=m.revision,
            file=m.filename,
            pattern="TRUNCATE in migration",
            message="TRUNCATE destroys all rows. Use DELETE with a WHERE clause instead.",
            severity="error",
        )]
    return []

config = MRTConfig(custom_checks=[check_no_truncate])

MRTFixture

The mrt pytest fixture. Obtained via the mrt fixture parameter — do not instantiate directly.

def test_migrations(mrt):
    mrt.assert_all_reversible()

Migration control

mrt.upgrade(revision="head")

Run alembic upgrade to the given revision.

mrt.upgrade("head")     # upgrade to latest
mrt.upgrade("001abc")   # upgrade to a specific revision

mrt.downgrade(revision="-1")

Run alembic downgrade by one step (or to a specific revision).

mrt.downgrade()         # roll back one step
mrt.downgrade("base")   # roll back to empty schema

Static analysis

mrt.check_static(versions_dir=None) → list[RiskWarning]

Run static analysis on migration files. Returns all detected risk warnings.

Parameter Default Description
versions_dir None Path to Alembic versions directory. Defaults to the path from alembic.ini. 
warnings = mrt.check_static()
errors = [w for w in warnings if w.severity == "error"]

mrt.assert_no_static_errors(versions_dir=None)

Fail the test if static analysis finds any error-severity risks.

def test_no_unsafe_migrations(mrt):
    mrt.assert_no_static_errors()

Rollback verification

mrt.check_revision(revision) → RevisionResult

Test a single revision for safe reversibility. The database must be at the state just before this revision when called.

Returns a RevisionResult (see below).

mrt.upgrade("001abc")
result = mrt.check_revision("002def")
assert result.passed

mrt.check_all() → list[RevisionResult]

Test every migration in the chain. Internally runs in O(n) upgrade operations.

results = mrt.check_all()
failed = [r for r in results if not r.passed]

mrt.assert_reversible(revision="head")

Assert that a single revision is safely reversible. Fails the test if not.

def test_latest_migration(mrt):
    mrt.assert_reversible()

mrt.assert_all_reversible()

Assert every migration in the chain is safely reversible. Prints a summary table and fails if any migration fails.

def test_all_migrations(mrt):
    mrt.assert_all_reversible()

Data integrity

mrt.seed(table, rows, pk_col="id")

Manually seed rows into a table (currently open schema).

mrt.upgrade("001abc")
mrt.seed("users", [{"id": 1, "name": "Alice"}])

mrt.assert_data_intact()

Assert that all previously seeded rows still exist and have their original values.

mrt.upgrade("head")
mrt.downgrade()
mrt.assert_data_intact()

mrt.reset()

Clear the internal seed state. Called automatically at fixture teardown.

RevisionResult

Return type of mrt.check_revision() and elements of mrt.check_all().

Attribute Type Description
revision str The Alembic revision ID
passed bool True if rollback was safe
skipped bool True if this revision is in MRTConfig.skip
skip_reason str The documented reason for skipping
failures list[str] Human-readable failure messages
risk_score int 0–100 risk score (25 per failure, capped at 100)

result.failure_summary() → str

Returns a formatted string of all failure messages.

result = mrt.check_revision("001abc")
if not result.passed:
    print(result.failure_summary())

RiskWarning

Returned by mrt.check_static() and by custom_checks functions.

Attribute Type Description
revision str Revision ID or filename stem
file str Migration filename
pattern str Short pattern name (e.g. "DROP COLUMN")
message str Human-readable explanation
severity str "error" or "warning"
line int | None Line number in the migration file

CLI Commands

mrt check <versions_dir>

Statically analyze migration files for rollback risks.

mrt check alembic/versions/
mrt check alembic/versions/ --strict
mrt check alembic/versions/ --format json
Option Default Description
--strict False Exit 1 on warnings as well as errors
--format / -f table Output format: table or json

Exit codes: 0 = safe, 1 = errors found (or warnings with --strict)

JSON output schema:

[
  {
    "revision": "001abc",
    "file": "001_create_users.py",
    "pattern": "DROP COLUMN",
    "severity": "error",
    "message": "op.drop_column('users', 'email') — data permanently lost...",
    "line": 12
  }
]

mrt fix <migration_file>

Suggest or apply a missing/broken downgrade() function.

mrt fix alembic/versions/001_create_users.py
mrt fix alembic/versions/001_create_users.py --apply
Option Default Description
--apply False Write the suggested fix to the file

mrt report <versions_dir>

Generate an HTML safety report for all migrations.

mrt report alembic/versions/
mrt report alembic/versions/ --output report.html
Option Default Description
--output / -o migration_report.html Output file path

mrt init

Scaffold conftest.py and test_migrations.py for your project. Auto-detects alembic.ini.

mrt init

mrt explain <migration_file>

Explain a migration in plain English using Claude AI.

mrt explain alembic/versions/001_create_users.py

Requires: pip install pytest-mrt[ai] and ANTHROPIC_API_KEY environment variable.

mrt version

Print the installed version.

mrt version