Merge pull request #751 from christianlouis/copilot/improve-database-migration-handling

feat(db): Add migration chain validation to CI, pre-commit, and developer docs
2026-03-17 10:55:10 +01:00
parent 0b0f43fa99 a007b4fd98
commit 7b7554decf
7 changed files with 853 additions and 1 deletions
@@ -44,6 +44,18 @@ jobs:
      - run: ruff check app/ tests/
      - run: ruff format --check app/ tests/

+  migration-chain:
+    name: Alembic Migration Chain Check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - name: Validate migration chain
+        run: python scripts/check_alembic_migrations.py
+
  html-lint:
    name: HTML Accessibility Lint
    runs-on: ubuntu-latest
@@ -138,7 +150,7 @@ jobs:
  build:
    name: Build & Push Docker Image
    runs-on: ubuntu-latest
-    needs: [run-tests, mypy, dependency-scan, html-lint]
+    needs: [run-tests, mypy, dependency-scan, html-lint, migration-chain]
    if: github.event_name == 'push'
    steps:
      - name: Checkout Code
@@ -48,6 +48,16 @@ repos:
            .env.demo
          )$

+  # Alembic migration chain validation
+  - repo: local
+    hooks:
+      - id: check-alembic-migrations
+        name: Check Alembic migration chain
+        entry: python scripts/check_alembic_migrations.py
+        language: python
+        pass_filenames: false
+        files: ^migrations/versions/.*\.py$
+
  # Conventional commits validation
  - repo: https://github.com/compilerla/conventional-pre-commit
    rev: v3.0.0
@@ -287,6 +287,27 @@ alembic revision --autogenerate -m "describe your change"

 Review the generated file in `migrations/versions/` before applying it.

+> **Tip:** For detailed guidance on naming conventions, idempotent patterns, parallel-branch workflows, and resolving merge conflicts, see the [Migration Workflow Guide](MigrationWorkflow.md).
+
+### Validating the Migration Chain
+
+A CI check and pre-commit hook validate that the migration chain has no broken
+references, duplicate revisions, or diverged heads.  Run the check locally:
+
+```bash
+python scripts/check_alembic_migrations.py
+python scripts/check_alembic_migrations.py --verbose   # extra detail
+```
+
+If you see **"Multiple migration heads detected"**, two branches added
+migrations from the same parent.  Create a merge migration:
+
+```bash
+alembic merge heads -m "merge_parallel_branches"
+```
+
+For a complete walk-through, see the [Migration Workflow Guide](MigrationWorkflow.md).
+
 ### Automating Migrations in Docker Compose

 Add a short-lived `migrate` service that runs before the API and Worker:
@@ -0,0 +1,360 @@
+# Migration Workflow
+
+This guide explains how to create, test, and merge Alembic database migrations in DocuElevate — especially when **multiple feature branches** add migrations in parallel.
+
+## Table of Contents
+
+- [Quick Reference](#quick-reference)
+- [Creating a New Migration](#creating-a-new-migration)
+- [Migration Naming Convention](#migration-naming-convention)
+- [Idempotent Migration Patterns](#idempotent-migration-patterns)
+- [Parallel Branch Development](#parallel-branch-development)
+- [Resolving Migration Conflicts](#resolving-migration-conflicts)
+- [CI Validation](#ci-validation)
+- [Pre-commit Hook](#pre-commit-hook)
+- [Troubleshooting](#troubleshooting)
+
+---
+
+## Quick Reference
+
+```bash
+# Create a new migration after editing app/models.py
+alembic revision --autogenerate -m "add_foobar_column"
+
+# Apply all pending migrations
+alembic upgrade head
+
+# Check current database version
+alembic current
+
+# View migration history
+alembic history --verbose
+
+# Detect multiple heads (diverged branches)
+alembic heads
+
+# Create a merge migration to resolve multiple heads
+alembic merge heads -m "merge_parallel_branches"
+
+# Validate migration chain integrity (CI script)
+python scripts/check_alembic_migrations.py
+python scripts/check_alembic_migrations.py --verbose
+```
+
+---
+
+## Creating a New Migration
+
+1. **Edit `app/models.py`** — add or modify SQLAlchemy model classes.
+
+2. **Generate the migration** from the repo root.  Use `--rev-id` to set the
+   revision identifier directly (avoids renaming afterwards):
+
+   ```bash
+   alembic revision --autogenerate --rev-id 037_add_my_new_table -m "add my new table"
+   ```
+
+   This creates `migrations/versions/037_add_my_new_table_add_my_new_table.py`
+   with `revision = "037_add_my_new_table"`.  Rename the file to match:
+
+   ```bash
+   mv migrations/versions/037_add_my_new_table_add_my_new_table.py \
+      migrations/versions/037_add_my_new_table.py
+   ```
+
+   Alternatively, generate with the default hash and then rename:
+
+   ```bash
+   alembic revision --autogenerate -m "add_my_new_table"
+   # Rename: mv migrations/versions/<hash>_add_my_new_table.py migrations/versions/037_add_my_new_table.py
+   # Update revision inside the file to match the filename stem.
+   ```
+
+   Alembic uses the `migrations/script.py.mako` template to generate the file.  The template includes inline comments about idempotent patterns — read them.
+
+3. **Review the generated code** — autogenerate is helpful but not perfect.  Check:
+   - Are new tables and columns detected correctly?
+   - Does the `downgrade()` reverse all changes?
+   - Are SQLite-incompatible operations wrapped in `batch_alter_table()`?
+
+4. **Test the migration** against a fresh database:
+
+   ```bash
+   # Apply
+   alembic upgrade head
+
+   # Rollback
+   alembic downgrade -1
+
+   # Re-apply
+   alembic upgrade head
+   ```
+
+5. **Run the chain validation**:
+
+   ```bash
+   python scripts/check_alembic_migrations.py
+   ```
+
+---
+
+## Migration Naming Convention
+
+All migration files follow a **sequential numeric prefix** scheme:
+
+```
+NNN_short_description.py
+```
+
+| Component | Rule |
+|-----------|------|
+| `NNN` | Three-digit zero-padded number, incrementing from the previous migration |
+| `short_description` | Lowercase snake_case summary of the change |
+
+The **`revision`** variable inside the file **must match the filename stem** exactly:
+
+```python
+# File: migrations/versions/037_add_classification_rules.py
+revision: str = "037_add_classification_rules"
+down_revision: Union[str, None] = "036_add_document_translation_fields"
+```
+
+The CI check (`scripts/check_alembic_migrations.py`) enforces this consistency.
+
+---
+
+## Idempotent Migration Patterns
+
+Migrations should be **idempotent** — safe to run even if the change already exists.  This is critical for SQLite compatibility and for recovering from partial failures.
+
+### Add a Column (only if missing)
+
+```python
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    if "my_table" in inspector.get_table_names():
+        existing = {c["name"] for c in inspector.get_columns("my_table")}
+        if "new_col" not in existing:
+            with op.batch_alter_table("my_table") as batch_op:
+                batch_op.add_column(sa.Column("new_col", sa.String(128), nullable=True))
+```
+
+### Create a Table (only if missing)
+
+```python
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    if "new_table" not in inspector.get_table_names():
+        op.create_table(
+            "new_table",
+            sa.Column("id", sa.Integer(), primary_key=True),
+            sa.Column("name", sa.String(255), nullable=False),
+        )
+```
+
+### Drop a Column (only if present)
+
+```python
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    if "my_table" in inspector.get_table_names():
+        existing = {c["name"] for c in inspector.get_columns("my_table")}
+        if "new_col" in existing:
+            with op.batch_alter_table("my_table") as batch_op:
+                batch_op.drop_column("new_col")
+```
+
+### Use `batch_alter_table` for SQLite
+
+SQLite does not support `ALTER TABLE DROP COLUMN` or `ALTER TABLE RENAME COLUMN` natively.  Alembic's `batch_alter_table` context manager works around this by recreating the table:
+
+```python
+with op.batch_alter_table("users") as batch_op:
+    batch_op.add_column(sa.Column("phone", sa.String(20), nullable=True))
+    batch_op.drop_column("fax")
+```
+
+---
+
+## Parallel Branch Development
+
+When two feature branches both add migrations from the same parent, the migration chain **diverges** into multiple heads.  This is normal and expected — Alembic supports it — but the heads must be merged before the code reaches `main`.
+
+### Example
+
+```
+main:   001 → 002 → 003
+                         ↘ Branch A: 004_add_widgets
+                         ↘ Branch B: 004_add_gadgets   ← two heads!
+```
+
+### How to Avoid Conflicts
+
+1. **Coordinate** — if two developers are both adding migrations, assign different sequence numbers (e.g., `037_` and `038_`).  Even if both depend on `036_`, different numbers prevent filename collisions.
+
+2. **Rebase early** — before opening a PR, rebase your branch onto the latest `main`:
+
+   ```bash
+   git fetch origin main
+   git rebase origin/main
+   ```
+
+   If `main` now has a new migration `037_*`, renumber yours to `038_*` and update `down_revision` to point at `037_*`.
+
+3. **Check for multiple heads** locally:
+
+   ```bash
+   python scripts/check_alembic_migrations.py
+   # or
+   alembic heads
+   ```
+
+---
+
+## Resolving Migration Conflicts
+
+If your PR's CI check reports **"Multiple migration heads detected"**, follow these steps:
+
+### Step 1 — Update Your Branch
+
+```bash
+git fetch origin main
+git merge origin/main
+# or
+git rebase origin/main
+```
+
+### Step 2 — Check Heads
+
+```bash
+python scripts/check_alembic_migrations.py --verbose
+```
+
+The output lists the conflicting heads.
+
+### Step 3 — Create a Merge Migration
+
+```bash
+alembic merge heads -m "merge_parallel_branches"
+```
+
+This generates a new migration with **two parents** (a merge point):
+
+```python
+down_revision = ("037_add_widgets", "037_add_gadgets")
+```
+
+### Step 4 — Rename and Validate
+
+Rename the merge migration to the next sequence number:
+
+```bash
+mv migrations/versions/<hash>_merge_parallel_branches.py \
+   migrations/versions/038_merge_parallel_branches.py
+```
+
+Update the `revision` inside to match, then validate:
+
+```bash
+python scripts/check_alembic_migrations.py
+```
+
+### Step 5 — Test
+
+```bash
+alembic upgrade head
+alembic downgrade -1
+alembic upgrade head
+```
+
+---
+
+## CI Validation
+
+The CI pipeline (`.github/workflows/ci.yml`) includes a **migration-chain** job that runs:
+
+```bash
+python scripts/check_alembic_migrations.py
+```
+
+This script checks for:
+
+| Check | Description |
+|-------|-------------|
+| Multiple heads | Diverged migration chains that need a merge migration |
+| Broken references | A `down_revision` that points to a non-existent revision |
+| Duplicate revisions | Two files declaring the same `revision` identifier |
+| Filename mismatches | The `revision` variable doesn't match the filename stem |
+
+The job runs in Stage 1 (fast-fail gates) alongside lint checks.  If it fails, the build is blocked until the migration chain is fixed.
+
+---
+
+## Pre-commit Hook
+
+A local pre-commit hook is configured in `.pre-commit-config.yaml` that runs the same check whenever you commit a change to `migrations/versions/`:
+
+```yaml
+- repo: local
+  hooks:
+    - id: check-alembic-migrations
+      name: Check Alembic migration chain
+      entry: python scripts/check_alembic_migrations.py
+      language: python
+      pass_filenames: false
+      files: ^migrations/versions/.*\.py$
+```
+
+Install the hook:
+
+```bash
+pip install pre-commit
+pre-commit install
+```
+
+---
+
+## Troubleshooting
+
+### "Multiple migration heads detected"
+
+See [Resolving Migration Conflicts](#resolving-migration-conflicts) above.
+
+### "Broken chain: revision X references down_revision Y which does not exist"
+
+You removed or renamed a migration that another migration depends on.  Either restore the missing file or update the dependent migration's `down_revision`.
+
+### "Filename mismatch: file declares revision=X but filename stem is Y"
+
+The `revision` string inside the Python file must match the filename (without `.py`).  Rename the file or update the variable.
+
+### "relation already exists" when running `alembic upgrade head`
+
+The database has a table that a pending migration tries to create.  Stamp the current state:
+
+```bash
+alembic stamp head
+```
+
+### Autogenerate doesn't detect my changes
+
+Ensure all models are imported in `migrations/env.py`.  The `from app.models import ...` block at the top must include your new model class.
+
+### SQLite "no such column" after downgrade
+
+SQLite has limited `ALTER TABLE` support.  Always use `op.batch_alter_table()` for column operations on existing tables.
+
+---
+
+## Further Reading
+
+- [Alembic Tutorial](https://alembic.sqlalchemy.org/en/latest/tutorial.html)
+- [Alembic Branch / Merge](https://alembic.sqlalchemy.org/en/latest/branches.html)
+- [Database Configuration Guide](DatabaseConfiguration.md)
@@ -0,0 +1,40 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+
+from typing import Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = ${repr(up_revision)}
+down_revision: Union[str, None] = ${repr(down_revision)}
+depends_on: Union[str, None] = None
+
+
+def upgrade() -> None:
+    """${message}."""
+    # Use ``op.batch_alter_table()`` for SQLite compatibility.
+    # Always check whether the table/column already exists before altering
+    # to keep migrations idempotent (safe to re-run).
+    #
+    # Example – add a column only if it is missing:
+    #
+    #     conn = op.get_bind()
+    #     inspector = sa.inspect(conn)
+    #     if "my_table" in inspector.get_table_names():
+    #         existing = {c["name"] for c in inspector.get_columns("my_table")}
+    #         if "new_col" not in existing:
+    #             with op.batch_alter_table("my_table") as batch_op:
+    #                 batch_op.add_column(sa.Column("new_col", sa.String(128), nullable=True))
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    """Reverse ${message}."""
+    ${downgrades if downgrades else "pass"}
@@ -0,0 +1,210 @@
+#!/usr/bin/env python3
+"""Validate Alembic migration chain integrity.
+
+This script checks the migration files in ``migrations/versions/`` for
+common problems that arise when multiple feature branches add migrations
+in parallel and then get merged into *main*.
+
+Checks performed
+~~~~~~~~~~~~~~~~
+1. **Multiple heads** – more than one migration without a child means the
+   chain has diverged and a merge migration is needed.
+2. **Broken down-revision references** – a migration points to a
+   ``down_revision`` that does not exist.
+3. **Duplicate revision IDs** – two files declare the same ``revision``.
+4. **Revision / filename mismatch** – the ``revision`` variable inside a
+   file does not match the stem of the filename (minus the numeric
+   prefix).
+
+Exit codes
+~~~~~~~~~~
+* **0** – all checks passed.
+* **1** – one or more problems detected (details printed to *stderr*).
+* **2** – unexpected runtime error.
+
+Usage::
+
+    python scripts/check_alembic_migrations.py          # from repo root
+    python scripts/check_alembic_migrations.py --verbose # extra detail
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import re
+import sys
+from pathlib import Path
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_REVISION_RE = re.compile(r'^revision\s*(?::\s*str\s*)?=\s*["\'](.+?)["\']', re.MULTILINE)
+_DOWN_REV_RE = re.compile(
+    r"^down_revision\s*(?::\s*Union\[str,\s*(?:None|tuple)\]\s*)?=\s*(.+)",
+    re.MULTILINE,
+)
+
+
+def _parse_down_revision(raw: str) -> list[str] | None:
+    """Parse a ``down_revision`` value into a list of parent revisions.
+
+    Returns ``None`` for the root migration (``down_revision = None``).
+    Returns a list with one or more strings otherwise.  Tuples are
+    returned for merge migrations (e.g. ``("017_a", "017_b")``).
+    """
+    # Strip inline comments (e.g. ``None  # type: ignore``)
+    raw = raw.strip()
+    if "#" in raw:
+        raw = raw[: raw.index("#")].strip()
+    try:
+        value = ast.literal_eval(raw)
+    except (ValueError, SyntaxError):
+        return [raw.strip("\"' ")]
+
+    if value is None:
+        return None
+    if isinstance(value, str):
+        return [value]
+    if isinstance(value, (tuple, list)):
+        return [str(v) for v in value]
+    return [str(value)]
+
+
+def _parse_migration(path: Path) -> dict | None:
+    """Extract ``revision`` and ``down_revision`` from a migration file."""
+    text = path.read_text(encoding="utf-8")
+
+    rev_match = _REVISION_RE.search(text)
+    down_match = _DOWN_REV_RE.search(text)
+
+    if not rev_match:
+        return None  # not a valid migration file
+
+    revision = rev_match.group(1)
+    down_revision = _parse_down_revision(down_match.group(1)) if down_match else None
+
+    return {
+        "path": path,
+        "revision": revision,
+        "down_revision": down_revision,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Checks
+# ---------------------------------------------------------------------------
+
+
+def check_migrations(versions_dir: Path, *, verbose: bool = False) -> list[str]:
+    """Run all migration-chain checks and return a list of error messages."""
+    errors: list[str] = []
+
+    # Collect all migrations ------------------------------------------------
+    migrations: dict[str, dict] = {}
+    py_files = sorted(versions_dir.glob("*.py"))
+    if not py_files:
+        errors.append(f"No migration files found in {versions_dir}")
+        return errors
+
+    for path in py_files:
+        if path.name == "__init__.py":
+            continue
+        info = _parse_migration(path)
+        if info is None:
+            if verbose:
+                print(f"  SKIP {path.name} (no revision found)", file=sys.stderr)
+            continue
+        rev = info["revision"]
+
+        # Check 1 – duplicate revision IDs
+        if rev in migrations:
+            errors.append(f"Duplicate revision '{rev}' in:\n  - {migrations[rev]['path'].name}\n  - {path.name}")
+        else:
+            migrations[rev] = info
+
+        if verbose:
+            parents = info["down_revision"] or ["(root)"]
+            print(f"  {rev}  ←  {', '.join(parents)}", file=sys.stderr)
+
+    # Build child map -------------------------------------------------------
+    all_revisions = set(migrations.keys())
+    children: dict[str, list[str]] = {rev: [] for rev in all_revisions}
+
+    for rev, info in migrations.items():
+        parents = info["down_revision"]
+        if parents is None:
+            continue
+        for parent in parents:
+            # Check 2 – broken down_revision references
+            if parent not in all_revisions:
+                errors.append(
+                    f"Broken chain: '{rev}' ({info['path'].name}) references "
+                    f"down_revision '{parent}' which does not exist."
+                )
+            else:
+                children[parent].append(rev)
+
+    # Check 3 – multiple heads (revisions with no children) -----------------
+    heads = [rev for rev, kids in children.items() if not kids]
+    if len(heads) > 1:
+        head_details = "\n".join(f"  - {h} ({migrations[h]['path'].name})" for h in sorted(heads))
+        errors.append(
+            f"Multiple migration heads detected ({len(heads)}).  "
+            f"Create a merge migration to resolve:\n{head_details}\n\n"
+            f'  Fix: alembic merge heads -m "merge_parallel_branches"'
+        )
+
+    # Check 4 – revision / filename consistency -----------------------------
+    for rev, info in migrations.items():
+        stem = info["path"].stem  # e.g. "017_add_pipelines"
+        if rev != stem:
+            errors.append(
+                f"Filename mismatch: file '{info['path'].name}' declares "
+                f"revision='{rev}' but filename stem is '{stem}'."
+            )
+
+    return errors
+
+
+# ---------------------------------------------------------------------------
+# CLI entry-point
+# ---------------------------------------------------------------------------
+
+
+def main(argv: list[str] | None = None) -> int:
+    """CLI entry-point.  Returns 0 on success, 1 on failure, 2 on error."""
+    parser = argparse.ArgumentParser(description="Check Alembic migration chain integrity.")
+    parser.add_argument(
+        "--versions-dir",
+        type=Path,
+        default=Path("migrations/versions"),
+        help="Path to Alembic versions directory (default: migrations/versions)",
+    )
+    parser.add_argument("--verbose", "-v", action="store_true", help="Print extra diagnostic info")
+    args = parser.parse_args(argv)
+
+    if not args.versions_dir.is_dir():
+        print(f"ERROR: versions directory not found: {args.versions_dir}", file=sys.stderr)
+        return 2
+
+    if args.verbose:
+        print("Scanning migrations…", file=sys.stderr)
+
+    errors = check_migrations(args.versions_dir, verbose=args.verbose)
+
+    if errors:
+        print(f"\n{'=' * 60}", file=sys.stderr)
+        print(f"  Migration chain problems found: {len(errors)}", file=sys.stderr)
+        print(f"{'=' * 60}\n", file=sys.stderr)
+        for i, err in enumerate(errors, 1):
+            print(f"  [{i}] {err}\n", file=sys.stderr)
+        return 1
+
+    print("✓ Migration chain is valid.", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,199 @@
+"""Tests for scripts/check_alembic_migrations.py."""
+
+# The script lives outside of the ``app`` package, so we import it by path.
+import importlib.util
+import textwrap
+from pathlib import Path
+
+import pytest
+
+_SCRIPT = Path(__file__).resolve().parent.parent / "scripts" / "check_alembic_migrations.py"
+_spec = importlib.util.spec_from_file_location("check_alembic_migrations", _SCRIPT)
+assert _spec and _spec.loader
+_mod = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_mod)  # type: ignore[union-attr]
+
+check_migrations = _mod.check_migrations
+main = _mod.main
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+def _write_migration(
+    directory: Path, filename: str, revision: str, down_revision: str | tuple[str, ...] | None
+) -> Path:
+    """Helper to create a minimal migration file."""
+    if down_revision is None:
+        down_rev_str = "None"
+    elif isinstance(down_revision, tuple):
+        down_rev_str = repr(down_revision)
+    else:
+        down_rev_str = f'"{down_revision}"'
+
+    content = textwrap.dedent(f'''\
+        """Test migration."""
+        from typing import Union
+        revision: str = "{revision}"
+        down_revision: Union[str, None] = {down_rev_str}
+        depends_on: Union[str, None] = None
+        def upgrade() -> None:
+            pass
+        def downgrade() -> None:
+            pass
+    ''')
+    path = directory / filename
+    path.write_text(content)
+    return path
+
+
+@pytest.fixture
+def versions_dir(tmp_path: Path) -> Path:
+    """Return a temporary versions directory."""
+    d = tmp_path / "versions"
+    d.mkdir()
+    return d
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+class TestCheckMigrations:
+    """Tests for the check_migrations function."""
+
+    def test_valid_linear_chain(self, versions_dir: Path) -> None:
+        """A simple linear chain should pass with no errors."""
+        _write_migration(versions_dir, "001_initial.py", "001_initial", None)
+        _write_migration(versions_dir, "002_add_col.py", "002_add_col", "001_initial")
+        _write_migration(versions_dir, "003_add_table.py", "003_add_table", "002_add_col")
+
+        errors = check_migrations(versions_dir)
+        assert errors == []
+
+    def test_valid_merge_migration(self, versions_dir: Path) -> None:
+        """A chain with a merge point should pass."""
+        _write_migration(versions_dir, "001_base.py", "001_base", None)
+        _write_migration(versions_dir, "002_a.py", "002_a", "001_base")
+        _write_migration(versions_dir, "002_b.py", "002_b", "001_base")
+
+        # Merge file with tuple down_revision
+        content = textwrap.dedent('''\
+            """Merge."""
+            from typing import Union
+            revision: str = "003_merge"
+            down_revision: Union[str, tuple] = ("002_a", "002_b")
+            depends_on: Union[str, None] = None
+            def upgrade() -> None:
+                pass
+            def downgrade() -> None:
+                pass
+        ''')
+        (versions_dir / "003_merge.py").write_text(content)
+
+        errors = check_migrations(versions_dir)
+        assert errors == []
+
+    def test_multiple_heads_detected(self, versions_dir: Path) -> None:
+        """Two unmerged branches should report multiple heads."""
+        _write_migration(versions_dir, "001_base.py", "001_base", None)
+        _write_migration(versions_dir, "002_a.py", "002_a", "001_base")
+        _write_migration(versions_dir, "002_b.py", "002_b", "001_base")
+
+        errors = check_migrations(versions_dir)
+        assert len(errors) == 1
+        assert "Multiple migration heads" in errors[0]
+        assert "002_a" in errors[0]
+        assert "002_b" in errors[0]
+
+    def test_broken_down_revision(self, versions_dir: Path) -> None:
+        """A migration pointing to a non-existent parent should be flagged."""
+        _write_migration(versions_dir, "001_base.py", "001_base", None)
+        _write_migration(versions_dir, "002_orphan.py", "002_orphan", "NONEXISTENT")
+
+        errors = check_migrations(versions_dir)
+        assert any("Broken chain" in e for e in errors)
+        assert any("NONEXISTENT" in e for e in errors)
+
+    def test_duplicate_revision(self, versions_dir: Path) -> None:
+        """Two files declaring the same revision should be flagged."""
+        _write_migration(versions_dir, "001_base.py", "001_base", None)
+        _write_migration(versions_dir, "002_first.py", "002_dup", "001_base")
+        _write_migration(versions_dir, "002_second.py", "002_dup", "001_base")
+
+        errors = check_migrations(versions_dir)
+        assert any("Duplicate revision" in e for e in errors)
+
+    def test_filename_mismatch(self, versions_dir: Path) -> None:
+        """A file whose revision doesn't match its filename should be flagged."""
+        _write_migration(versions_dir, "001_base.py", "001_base", None)
+        # filename stem is "002_wrong_name" but revision says "002_correct_name"
+        _write_migration(versions_dir, "002_wrong_name.py", "002_correct_name", "001_base")
+
+        errors = check_migrations(versions_dir)
+        assert any("Filename mismatch" in e for e in errors)
+
+    def test_empty_directory(self, versions_dir: Path) -> None:
+        """An empty versions directory should report an error."""
+        errors = check_migrations(versions_dir)
+        assert len(errors) == 1
+        assert "No migration files found" in errors[0]
+
+    def test_init_py_is_skipped(self, versions_dir: Path) -> None:
+        """__init__.py files should be ignored."""
+        (versions_dir / "__init__.py").write_text("")
+        _write_migration(versions_dir, "001_base.py", "001_base", None)
+
+        errors = check_migrations(versions_dir)
+        assert errors == []
+
+    def test_non_migration_file_skipped(self, versions_dir: Path) -> None:
+        """A .py file without a revision variable should be silently skipped."""
+        (versions_dir / "helper.py").write_text("# just a helper\nx = 1\n")
+        _write_migration(versions_dir, "001_base.py", "001_base", None)
+
+        errors = check_migrations(versions_dir)
+        assert errors == []
+
+
+@pytest.mark.unit
+class TestMainCLI:
+    """Tests for the CLI entry-point."""
+
+    def test_success_returns_zero(self, versions_dir: Path) -> None:
+        """Valid chain should exit 0."""
+        _write_migration(versions_dir, "001_base.py", "001_base", None)
+        rc = main(["--versions-dir", str(versions_dir)])
+        assert rc == 0
+
+    def test_failure_returns_one(self, versions_dir: Path) -> None:
+        """Invalid chain should exit 1."""
+        _write_migration(versions_dir, "001_base.py", "001_base", None)
+        _write_migration(versions_dir, "002_a.py", "002_a", "001_base")
+        _write_migration(versions_dir, "002_b.py", "002_b", "001_base")
+
+        rc = main(["--versions-dir", str(versions_dir)])
+        assert rc == 1
+
+    def test_missing_directory_returns_two(self, tmp_path: Path) -> None:
+        """Non-existent versions directory should exit 2."""
+        rc = main(["--versions-dir", str(tmp_path / "does_not_exist")])
+        assert rc == 2
+
+    def test_verbose_flag(self, versions_dir: Path) -> None:
+        """The --verbose flag should not crash."""
+        _write_migration(versions_dir, "001_base.py", "001_base", None)
+        rc = main(["--versions-dir", str(versions_dir), "--verbose"])
+        assert rc == 0
+
+    def test_real_migrations(self) -> None:
+        """Smoke test against the actual project migrations."""
+        real_dir = Path(__file__).resolve().parent.parent / "migrations" / "versions"
+        if not real_dir.is_dir():
+            pytest.skip("migrations/versions directory not found in working tree")
+        rc = main(["--versions-dir", str(real_dir)])
+        assert rc == 0