feat: built-in default tests + schema drift detection (v1.1.0) (#27)

* feat: add built-in default tests and schema drift detection (v1.1.0)

- Auto-collect 5 built-in tests when MRTConfig is registered (no imports needed).
  Disable with mrt_default_tests = "false" in pytest.ini.
- Add MRTFixture.assert_schema_matches() — fails on SQLAlchemy model/migration drift.
  Django mode delegates to manage.py makemigrations --check.
- Add mrt drift CLI command for schema drift reporting.
- Add MRTConfig.target_metadata field for import-path based metadata resolution.
- Fix get_versions_dir() to return actual versions/ path, not script root.

* fix: resolve 6 code-review bugs (drift safety, bare excepts, DB state)

* style: apply ruff format

* fix: narrow Item | Collector to Item in pytest_collection_modifyitems

Author: croc100

CHANGELOG.md

@@ -7,6 +7,16 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
 
 ---
 
+## [1.1.0] — 2026-06-07
+
+### Added
+- **Built-in default tests**: Five tests are now auto-collected when `MRTConfig` is registered in `conftest.py` — no imports required. Tests: `test_mrt_single_head`, `test_mrt_upgrade`, `test_mrt_downgrade_base`, `test_mrt_static_no_errors`, `test_mrt_schema_matches_models`. Disable with `mrt_default_tests = "false"` in `pytest.ini` / `pyproject.toml`.
+- **Schema drift detection**: `MRTFixture.assert_schema_matches()` — fails if the DB schema after running all migrations does not match the SQLAlchemy model definitions. Accepts a `MetaData` instance or an import-path string (`"myapp.models:Base"`). Django mode delegates to `manage.py makemigrations --check`.
+- **`mrt drift` CLI command**: `mrt drift myapp.models:Base --config alembic.ini --db-url sqlite:///test.db` — runs migrations to head, compares schema against models, prints a diff table, exits 1 on drift.
+- **`MRTConfig.target_metadata`**: New field (`str | None`) — import path for the SQLAlchemy `Base` or `MetaData` used by `assert_schema_matches()` and `test_mrt_schema_matches_models`.
+
+---
+
 ## [1.0.1] — 2026-06-06
 
 ### Fixed

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "pytest-mrt"
-version = "1.0.1"
+version = "1.1.0"
 description = "Catch database migration rollback failures before they reach production"
 readme = "README.md"
 license = { file = "LICENSE" }

pytest_mrt/cli.py

@@ -126,6 +126,77 @@ def check(
         raise typer.Exit(1)
 
 
+# ──────────────────────────────────────────────
+# mrt drift
+# ──────────────────────────────────────────────
+
+
+@app.command("drift")
+def drift(
+    metadata: str = typer.Argument(
+        help="SQLAlchemy metadata import path, e.g. 'myapp.models:Base'"
+    ),
+    alembic_ini: str = typer.Option("alembic.ini", "--config", "-c", help="Path to alembic.ini"),
+    db_url: str = typer.Option("", "--db-url", help="Database URL (overrides alembic.ini)"),
+) -> None:
+    """Check if SQLAlchemy model definitions match the current migration state.
+
+    Compares the live DB schema (after running all migrations) against the
+    SQLAlchemy models you point at. Exits 0 if clean, 1 if drift is found.
+
+    Example:
+
+        mrt drift myapp.models:Base --config alembic.ini --db-url sqlite:///test.db
+    """
+    from .core.drift import compare_schema, describe_diff, load_metadata
+    from .core.runner import MigrationRunner
+
+    # Load metadata
+    try:
+        target_metadata = load_metadata(metadata)
+    except (ValueError, ImportError, AttributeError) as exc:
+        console.print(f"[red]Error loading metadata:[/red] {exc}")
+        raise typer.Exit(1)
+
+    # Build runner (needs alembic.ini + db_url)
+    if not Path(alembic_ini).exists():
+        console.print(f"[red]alembic.ini not found:[/red] {alembic_ini}")
+        raise typer.Exit(1)
+
+    try:
+        runner = MigrationRunner(alembic_ini, db_url)
+    except Exception as exc:
+        console.print(f"[red]Failed to connect:[/red] {exc}")
+        raise typer.Exit(1)
+
+    console.print("[dim]Upgrading to head...[/dim]")
+    try:
+        runner.upgrade("head")
+    except Exception as exc:
+        console.print(f"[red]Migration failed:[/red] {exc}")
+        raise typer.Exit(1)
+
+    console.print("[dim]Comparing schema...[/dim]")
+    diffs = compare_schema(runner.engine, target_metadata)
+
+    if not diffs:
+        console.print("[green]✓ No schema drift — models match migrations.[/green]")
+        raise typer.Exit(0)
+
+    table = Table(box=box.ROUNDED, title="Schema Drift", show_lines=True)
+    table.add_column("#", style="dim", no_wrap=True)
+    table.add_column("Difference")
+
+    for i, d in enumerate(diffs, 1):
+        table.add_row(str(i), describe_diff(d))
+
+    console.print(table)
+    console.print(
+        f"\n[red]{len(diffs)} difference(s) found.[/red] Run migrations or update your models."
+    )
+    raise typer.Exit(1)
+
+
 # ──────────────────────────────────────────────
 # mrt init
 # ──────────────────────────────────────────────

pytest_mrt/config.py

@@ -76,3 +76,9 @@ class MRTConfig:
     # Model used by `mrt explain`. Defaults to DEFAULT_EXPLAIN_MODEL.
     # Override to use a different Claude model, e.g. "claude-3-5-haiku-latest".
     explain_model: str = DEFAULT_EXPLAIN_MODEL
+
+    # Import path for the SQLAlchemy declarative Base (or MetaData) used by
+    # assert_schema_matches() and the built-in test_mrt_schema_matches_models test.
+    # Format: "myapp.models:Base" or "myapp.models:Base.metadata"
+    # Example: target_metadata="myproject.db.models:Base"
+    target_metadata: str | None = None

pytest_mrt/core/drift.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+
+def load_metadata(metadata_path: str):
+    """Import SQLAlchemy metadata from a dotted path like 'myapp.models:Base'.
+
+    The format is 'module.path:AttributeName'. Dotted attribute traversal after
+    the colon is not supported — use 'myapp.models:Base' (not 'myapp.models:Base.metadata').
+    Both a declarative Base class and a MetaData instance are accepted.
+    """
+    if ":" not in metadata_path:
+        raise ValueError(
+            f"Invalid metadata path '{metadata_path}'. Use the form 'myapp.models:Base'."
+        )
+    module_path, attr = metadata_path.rsplit(":", 1)
+    if "." in attr:
+        raise ValueError(
+            f"Dotted attribute '{attr}' is not supported after the colon. "
+            f"Use 'myapp.models:Base' instead of 'myapp.models:Base.metadata'."
+        )
+    import importlib
+
+    mod = importlib.import_module(module_path)
+    obj = getattr(mod, attr)
+    # Accept either a declarative Base class or a MetaData instance directly.
+    return getattr(obj, "metadata", obj)
+
+
+def compare_schema(engine, target_metadata) -> list:
+    """Return alembic autogenerate diffs between DB schema and target_metadata."""
+    from alembic.autogenerate import compare_metadata
+    from alembic.runtime.migration import MigrationContext
+
+    with engine.connect() as conn:
+        ctx = MigrationContext.configure(conn)
+        return compare_metadata(ctx, target_metadata)
+
+
+def describe_diff(diff) -> str:
+    """Format a single alembic autogenerate diff tuple into a human-readable string."""
+    if not isinstance(diff, tuple):
+        return repr(diff)
+    kind = diff[0]
+    if kind == "add_table":
+        return f"add table '{diff[1].name}' (in models, missing from DB)"
+    if kind == "remove_table":
+        return f"remove table '{diff[1].name}' (in DB, missing from models)"
+    if kind == "add_column":
+        return f"add column '{diff[2]}.{diff[3].name}' (in models, missing from DB)"
+    if kind == "remove_column":
+        return f"remove column '{diff[2]}.{diff[3].name}' (in DB, missing from models)"
+    if kind == "modify_type":
+        return f"column '{diff[2]}.{diff[3]}' type mismatch: models={diff[5]!r}, DB={diff[4]!r}"
+    if kind == "modify_nullable":
+        return f"column '{diff[2]}.{diff[3]}' nullable mismatch: models={diff[5]}, DB={diff[4]}"
+    return repr(diff)

pytest_mrt/core/runner.py

@@ -72,4 +72,8 @@ def get_revisions(self) -> list:
 
     def get_versions_dir(self) -> str:
         script = ScriptDirectory.from_config(self.alembic_cfg)
-        return script.dir
+        # script.versions returns the single version_locations path (where .py migration files live).
+        # script.dir is the script_location root (contains env.py, script.py.mako, etc.) — not what we want.
+        # script.versions raises CommandError when multiple version_locations are configured; surface
+        # that instead of silently falling back to script.dir (which contains non-migration files).
+        return script.versions

pytest_mrt/default_tests.py

@@ -0,0 +1,76 @@
+"""
+Built-in default tests for pytest-mrt.
+
+These are automatically collected when MRTConfig is registered in conftest.py.
+To disable auto-collection, add the following to pyproject.toml or pytest.ini:
+
+    [tool.pytest.ini_options]
+    mrt_default_tests = "false"
+
+You can also import individual tests explicitly:
+
+    from pytest_mrt.default_tests import test_mrt_upgrade, test_mrt_downgrade_base
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+def test_mrt_single_head(mrt) -> None:
+    """Exactly one head revision exists in the migration chain."""
+    if mrt._django_mode:
+        pytest.skip("single-head check not applicable to Django mode")
+    from alembic.script import ScriptDirectory
+
+    script = ScriptDirectory.from_config(mrt._runner.alembic_cfg)
+    heads = script.get_heads()
+    assert len(heads) == 1, (
+        f"Expected a single head revision, found {len(heads)}: {heads}.\n"
+        "Run `alembic merge heads` to create a merge migration."
+    )
+
+
+def test_mrt_upgrade(mrt) -> None:
+    """Migration chain upgrades to head without error."""
+    if mrt._django_mode:
+        pytest.skip("use test_mrt_all_reversible for Django mode")
+    mrt.upgrade("head")
+
+
+def test_mrt_downgrade_base(mrt) -> None:
+    """Migration chain downgrades to base without error."""
+    if mrt._django_mode:
+        pytest.skip("use test_mrt_all_reversible for Django mode")
+    mrt.upgrade("head")
+    mrt._runner.downgrade_base()
+    try:
+        mrt.upgrade("head")  # restore to head so subsequent tests start clean
+    except Exception as exc:
+        pytest.fail(
+            f"Migration chain failed to re-upgrade to head after downgrade_base: {exc}\n"
+            "DB state is now at 'base'. Fix the upgrade() failure before continuing."
+        )
+
+
+def test_mrt_static_no_errors(mrt) -> None:
+    """No static analysis errors found in migration files."""
+    mrt.assert_no_static_errors()
+
+
+def test_mrt_schema_matches_models(mrt) -> None:
+    """Model definitions match the migration state (no schema drift).
+
+    Requires ``MRTConfig(target_metadata='myapp.models:Base')`` to be set.
+    Skipped automatically when target_metadata is not configured.
+    """
+    if mrt._config.target_metadata is None:
+        pytest.skip(
+            "target_metadata not configured — skipping schema drift check.\n"
+            "Set MRTConfig(target_metadata='myapp.models:Base') to enable."
+        )
+    # Ensure migrations are applied before comparing schema.
+    # This makes the test safe to run in isolation (e.g. pytest -k test_mrt_schema_matches_models).
+    if not mrt._django_mode:
+        mrt.upgrade("head")
+    mrt.assert_schema_matches()

pytest_mrt/plugin.py

@@ -6,6 +6,7 @@
 
 from .config import MRTConfig
 from .core.detector import RiskWarning, analyze_migrations
+from .core.drift import compare_schema, describe_diff, load_metadata
 from .core.runner import MigrationRunner
 from .core.schema import SchemaSnapshot
 from .core.seeder import SmartSeeder
@@ -226,14 +227,112 @@ def assert_all_reversible(self, apps: list[str] | None = None) -> None:
                 lines.append(r.failure_summary())
             pytest.fail("Some migrations are not safely reversible:\n" + "\n".join(lines))
 
+    # ── schema drift ──────────────────────────────────────────────────
+
+    def assert_schema_matches(
+        self,
+        target_metadata=None,
+        metadata_path: str | None = None,
+    ) -> None:
+        """Fail if the DB schema does not match the SQLAlchemy model definitions.
+
+        For Django mode, delegates to ``manage.py makemigrations --check``.
+
+        Args:
+            target_metadata: A SQLAlchemy ``MetaData`` instance (or declarative
+                ``Base``) to compare against.  When omitted, falls back to
+                ``MRTConfig.target_metadata`` (an import-path string).
+            metadata_path: Import path override, e.g. ``"myapp.models:Base"``.
+                Takes precedence over ``MRTConfig.target_metadata``.
+        """
+        if self._django_mode:
+            self._assert_django_no_drift()
+            return
+
+        if target_metadata is None:
+            path = metadata_path or self._config.target_metadata
+            if path is None:
+                raise ValueError(
+                    "assert_schema_matches() requires either a target_metadata argument "
+                    "or MRTConfig(target_metadata='myapp.models:Base')."
+                )
+            target_metadata = load_metadata(path)
+
+        diffs = compare_schema(self._runner.engine, target_metadata)
+        if diffs:
+            lines = [f"  {describe_diff(d)}" for d in diffs]
+            pytest.fail(
+                f"Schema drift detected ({len(diffs)} difference(s)):\n" + "\n".join(lines)
+            )
+
+    def _assert_django_no_drift(self) -> None:
+        from io import StringIO
+
+        from django.core.management import call_command
+
+        out = StringIO()
+        try:
+            call_command("makemigrations", "--check", "--dry-run", stdout=out, stderr=out)
+        except SystemExit as exc:
+            if exc.code != 0:
+                pytest.fail(
+                    "Schema drift: model changes detected that don't have migrations.\n"
+                    "Run `python manage.py makemigrations` to generate them."
+                )
+        except Exception as exc:
+            pytest.fail(
+                f"assert_schema_matches() failed while checking Django migrations: {exc}\n"
+                "Check that DJANGO_SETTINGS_MODULE is set correctly and all models can be imported."
+            )
+
     def reset(self) -> None:
         self._seeder.reset()
 
 
+def pytest_addoption(parser: pytest.Parser) -> None:
+    parser.addini(
+        "mrt_default_tests",
+        help="Set to 'false' to disable auto-collected built-in MRT tests.",
+        default="true",
+    )
+
+
 def pytest_configure(config: pytest.Config) -> None:
     config.addinivalue_line("markers", "mrt: migration rollback test")
 
 
+def pytest_collection_modifyitems(
+    session: pytest.Session,
+    config: pytest.Config,
+    items: list[pytest.Item],
+) -> None:
+    """Prepend built-in default tests when MRTConfig is registered."""
+    if getattr(config, "_mrt_config", None) is None:
+        return
+    if config.getini("mrt_default_tests") == "false":
+        return
+
+    from pathlib import Path
+
+    try:
+        from _pytest.python import Module as _PytestModule
+
+        import pytest_mrt.default_tests as _dt
+
+        dt_path = Path(_dt.__file__)
+        module = _PytestModule.from_parent(session, path=dt_path)
+        new_items: list[pytest.Item] = [i for i in module.collect() if isinstance(i, pytest.Item)]
+        items[:0] = new_items
+    except Exception as _exc:
+        import warnings
+
+        warnings.warn(
+            f"pytest-mrt: failed to inject built-in default tests: {_exc}\n"
+            "Set mrt_default_tests = 'false' in pytest.ini to suppress this warning.",
+            stacklevel=2,
+        )
+
+
 @pytest.fixture
 def mrt(request: pytest.FixtureRequest) -> Iterator[MRTFixture]:
     cfg: MRTConfig = getattr(request.config, "_mrt_config", None) or MRTConfig()

tests/test_plugin.py

@@ -523,6 +523,8 @@ def test_fixture_type(mrt):
         def test_fixture_has_config(mrt):
             assert mrt._config is not None
     """)
+    # Disable default test injection so this test focuses on fixture wiring only.
+    pytester.makeini("[pytest]\nmrt_default_tests = false\n")
     result = pytester.runpytest("-v")
     result.assert_outcomes(passed=2)