test: fill coverage completeness gaps (seed E2E, seed_custom, Django mode) (#87)

* fix: seed() crash in Django mode and silent seeder failures

- seed() used self._runner.engine which is None in Django mode;
  switch to self._seeder.engine which is always set
- replace bare except/pass in SmartSeeder.seed_table() with
  warnings.warn so failures are visible instead of swallowed silently

Closes #83, #84

* fix: seed() ignored rows param, UNIQUE collisions in check_all, silent custom-seed failures

Three root causes fixed:

1. seed() was calling seed_table() (auto-generates rows) instead of
   inserting the caller-supplied rows. Add SmartSeeder.seed_custom()
   and wire seed() through it so the rows parameter is actually used.

2. SmartSeeder.seed_table() used a deterministic hash for generated
   values, causing UNIQUE constraint violations when check_all() seeded
   the same table across consecutive revision checks (prior rows survive
   downgrade and remain in the DB). Add a per-instance random offset so
   each SmartSeeder produces distinct values.

3. Custom-seed insertion in verifier._build_seeder() silently swallowed
   all exceptions. Replace bare except/pass with warnings.warn so
   failures are visible.

* fix: seeder value generation — prevent UNIQUE collisions for any column type

String columns used a fixed-width row_index format that truncates to the
same prefix when the instance offset is large (e.g. offset=500000 →
"mrt_title_500000"[:10] == "mrt_title_" for all rows). Switch to
uuid4().hex[:limit] which is unique regardless of column length.

Datetime/date/time types had a narrow modulo range (28 days, 24 hours)
causing cross-instance collisions. Spread values over a 10-year window
using the full hash seed instead.

* feat: add PostgreSQL concurrency safety patterns MRT211-MRT213

Three new static analysis checks for lock-heavy PostgreSQL operations:

MRT211 — create_foreign_key() without postgresql_not_valid=True
  Acquires ShareRowExclusiveLock and scans the referencing table.
  Safe path: add with NOT VALID, then VALIDATE CONSTRAINT separately.

MRT212 — create_check_constraint() without postgresql_not_valid=True
  Full table scan under AccessShareLock. Same two-step approach applies.

MRT213 — alter_column(nullable=False) on an existing column
  SET NOT NULL triggers a full table scan under AccessExclusiveLock.
  Safe path on PG 12+: validated CHECK constraint first, then SET NOT NULL.

Closes #85

* revert: remove MRT211/MRT212 (lock warnings outside core scope)

MRT211 (FK without NOT VALID) and MRT212 (CHECK without NOT VALID) warn
about PostgreSQL lock duration — a performance/availability concern, not
a data survival or rollback safety concern. That's django-migration-linter
territory.

MRT213 (SET NOT NULL on existing column) stays: it's about existing NULL
data causing migration failure, consistent with MRT401/MRT403.

* test: fill coverage gaps — seed() E2E, seed_custom(), Django mode seed(), cli drift

* fix: don't track custom seed rows whose INSERT failed (false positive)

A failed custom-seed INSERT was caught and warned, but the row had
already been appended to the seeder's tracked rows. verify() then
reported it as 'lost after rollback', blaming the migration for data
the seed never inserted. Track rows only after a successful insert.

Also applies ruff format to verifier.py (fixes CI lint failure).

Author: croc100

pytest_mrt/core/detector.py

@@ -773,6 +773,39 @@ def _check_create_type_without_drop(m: MigrationAST) -> list[RiskWarning]:
     ]
 
 
+def _check_set_not_null_alter_column(m: MigrationAST) -> list[RiskWarning]:
+    """
+    alter_column(nullable=False) issues SET NOT NULL which does a full table
+    scan on PostgreSQL to verify no NULLs exist. Holds AccessExclusiveLock
+    for the scan duration.
+
+    Safe path (PostgreSQL 12+): add a NOT NULL check constraint with NOT VALID,
+    validate it, then ALTER COLUMN SET NOT NULL (PostgreSQL skips the scan when
+    a validated constraint already covers it).
+    """
+    warnings = []
+    for c in m.upgrade_calls():
+        if c.method == "alter_column":
+            nullable = m.kwarg_bool(c.node, "nullable")
+            if nullable is False:
+                table = m.str_arg(c.node, 0) or "?"
+                col = m.str_arg(c.node, 1) or "?"
+                warnings.append(
+                    _warn(
+                        m,
+                        "SET NOT NULL on existing column",
+                        f"alter_column('{table}', '{col}', nullable=False) runs SET NOT NULL — "
+                        "full table scan with AccessExclusiveLock on PostgreSQL. "
+                        "Safe path on PG 12+: add a NOT NULL check constraint (NOT VALID), "
+                        "validate it, then SET NOT NULL.",
+                        "warning",
+                        line=c.node.lineno,
+                        code="MRT213",
+                    )
+                )
+    return warnings
+
+
 _PER_FILE_CHECKS = [
     _check_batch_alter_drop,  # first: batch context needs special handling
     _check_downgrade_exists,
@@ -803,6 +836,7 @@ def _check_create_type_without_drop(m: MigrationAST) -> list[RiskWarning]:
     _check_drop_foreign_key,
     _check_create_trigger_without_drop,
     _check_create_type_without_drop,
+    _check_set_not_null_alter_column,
 ]
 
 

pytest_mrt/core/seeder.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+import random
 import re
 import uuid
+import warnings
 from dataclasses import dataclass
-from datetime import date, datetime, time
+from datetime import date, datetime, time, timedelta
 from decimal import Decimal
 from typing import Any
 
@@ -121,18 +123,20 @@ def _generate_value(
     if any(x in t for x in ("BYTEA", "VARBINARY", "BLOB", "BINARY")):
         return f"mrt_{row_index}".encode()
     if "TIMESTAMP" in t or "DATETIME" in t:
-        return datetime(2024, 1, row_index % 28 + 1, row_index % 24, 0, 0)
+        return datetime(2020, 1, 1) + timedelta(seconds=seed % (10 * 365 * 86400))
     if "DATE" in t:
-        return date(2024, 1, row_index % 28 + 1)
+        return date(2020, 1, 1) + timedelta(days=seed % (10 * 365))
     if "TIME" in t:
-        return time(row_index % 24, 0, 0)
+        return time((seed // 3600) % 24, (seed // 60) % 60, seed % 60)
     if any(x in t for x in ("VARCHAR", "TEXT", "CHAR", "STRING", "CLOB")):
         m = re.search(r"\((\d+)\)", t)
         limit = int(m.group(1)) if m else 255
-        val = f"mrt_{col.name[:8]}_{row_index:04d}"
-        return val[:limit]
+        # UUID guarantees uniqueness regardless of column length or row_index magnitude,
+        # avoiding UNIQUE collisions when the same table is seeded across multiple
+        # check_revision() calls (prior rows survive downgrade and remain in the DB).
+        return uuid.uuid4().hex[:limit]
 
-    return f"mrt_{row_index}"
+    return uuid.uuid4().hex
 
 
 def _normalize_for_compare(val: Any) -> Any:
@@ -207,6 +211,10 @@ class SmartSeeder:
     def __init__(self, engine: Engine):
         self.engine = engine
         self._rows: list[SeededRow] = []
+        # Random offset makes generated values unique per seeder instance, preventing
+        # UNIQUE constraint collisions when check_all() seeds the same table multiple times
+        # (prior check's rows survive downgrade and remain in the DB for the next check).
+        self._offset = random.randint(0, 10**6)
 
     def _q(self, name: str) -> str:
         return _q(self.engine, name)
@@ -248,7 +256,7 @@ def seed_table(self, table: TableInfo, count: int = 3) -> None:
                     enum_vals = None
                     if "ENUM" in col_info.type_str.upper():
                         enum_vals = _get_enum_values(self.engine, table.name, col_name)
-                    val = _generate_value(col_info, row_index, enum_vals)
+                    val = _generate_value(col_info, self._offset + row_index, enum_vals)
                     if val is not None:
                         row[col_name] = val
                 # nullable columns left as NULL intentionally
@@ -297,8 +305,46 @@ def seed_table(self, table: TableInfo, count: int = 3) -> None:
                         full_row = dict(result.mappings().first() or {})
                         self._rows.append(SeededRow(table.name, pk_col, pk_val, full_row))
 
-            except Exception:
-                pass
+            except Exception as exc:
+                warnings.warn(
+                    f"pytest-mrt: failed to seed row {row_index} into '{table.name}': {exc}",
+                    stacklevel=2,
+                )
+
+    def seed_custom(self, table: str, pk_col: str, rows: list[dict]) -> None:
+        """Insert caller-supplied rows and track them for verify()."""
+        tq = self._q(table)
+        pkq = self._q(pk_col)
+        for row in rows:
+            cols = ", ".join(self._q(c) for c in row)
+            placeholders = ", ".join(f":p_{c}" for c in row)
+            params = {f"p_{c}": v for c, v in row.items()}
+            stmt = text(f"INSERT INTO {tq} ({cols}) VALUES ({placeholders})")
+            try:
+                with self.engine.begin() as conn:
+                    conn.execute(stmt, params)
+            except Exception as exc:
+                warnings.warn(
+                    f"pytest-mrt: failed to insert custom row into '{table}': {exc}",
+                    stacklevel=2,
+                )
+                continue
+
+            pk_val = row.get(pk_col)
+            if pk_val is None:
+                with self.engine.connect() as conn:
+                    result = conn.execute(
+                        text(f"SELECT {pkq} FROM {tq} ORDER BY {pkq} DESC LIMIT 1")
+                    )
+                    pk_val = result.scalar()
+
+            if pk_val is not None:
+                with self.engine.connect() as conn:
+                    result = conn.execute(
+                        text(f"SELECT * FROM {tq} WHERE {pkq} = :pk"), {"pk": pk_val}
+                    )
+                    full_row = dict(result.mappings().first() or {})
+                    self._rows.append(SeededRow(table, pk_col, pk_val, full_row))
 
     def verify(self) -> list[str]:
         """

pytest_mrt/core/verifier.py

@@ -63,10 +63,6 @@ def _build_seeder(self, schema: SchemaSnapshot) -> SmartSeeder:
             if tname in self.custom_seeds:
                 rows = self.custom_seeds[tname]()
                 pk_col = table_info.pk_cols[0] if table_info.pk_cols else "id"
-                for row in rows:
-                    seeder._rows.append(
-                        SeededRow(table=tname, pk_col=pk_col, pk_val=row.get(pk_col), data=row)
-                    )
                 for row in rows:
                     # Use dialect-aware quoting — fixes MySQL double-quote bug
                     def q(name: str) -> str:
@@ -83,8 +79,21 @@ def q(name: str) -> str:
                                 text(f"INSERT INTO {q(tname)} ({cols}) VALUES ({placeholders})"),
                                 params,
                             )
-                    except Exception:
-                        pass
+                    except Exception as exc:
+                        import warnings
+
+                        warnings.warn(
+                            f"pytest-mrt: failed to insert custom seed row into '{tname}': {exc}",
+                            stacklevel=2,
+                        )
+                        # Insert failed — do NOT track this row, otherwise verify()
+                        # would report it as "lost after rollback" (false positive).
+                        continue
+
+                    # Only track rows that were actually inserted.
+                    seeder._rows.append(
+                        SeededRow(table=tname, pk_col=pk_col, pk_val=row.get(pk_col), data=row)
+                    )
             else:
                 seeder.seed_table(table_info)
         return seeder

pytest_mrt/plugin.py

@@ -112,11 +112,10 @@ def downgrade(self, revision: str = "-1") -> None:
     # ── manual seeding ────────────────────────────────────────────────
 
     def seed(self, table: str, rows: list[dict], pk_col: str = "id") -> None:
-        snap = SchemaSnapshot.capture(self._runner.engine)
-        if table in snap.tables:
-            self._seeder.seed_table(snap.tables[table])
-        else:
+        snap = SchemaSnapshot.capture(self._seeder.engine)
+        if table not in snap.tables:
             raise ValueError(f"Table '{table}' not found in current schema")
+        self._seeder.seed_custom(table, pk_col, rows)
 
     # ── static analysis ───────────────────────────────────────────────
 

tests/test_cli.py

@@ -739,3 +739,173 @@ def downgrade():
     )
     result = runner.invoke(app, ["fix", str(f)])
     assert "--apply" in result.output
+
+
+# ── mrt drift ────────────────────────────────────────────────────────
+
+
+def _alembic_env(tmp_path):
+    """Minimal Alembic setup returning (ini_path, versions_dir, db_url)."""
+    import textwrap as _tw
+
+    db = tmp_path / "test.db"
+    versions = tmp_path / "versions"
+    versions.mkdir()
+    env_py = tmp_path / "env.py"
+    env_py.write_text(
+        _tw.dedent("""
+        from alembic import context
+        from sqlalchemy import engine_from_config, pool
+
+        config = context.config
+
+        def run_migrations_offline():
+            url = config.get_main_option("sqlalchemy.url")
+            context.configure(url=url, literal_binds=True)
+            with context.begin_transaction():
+                context.run_migrations()
+
+        def run_migrations_online():
+            connectable = engine_from_config(
+                config.get_section(config.config_ini_section),
+                prefix="sqlalchemy.",
+                poolclass=pool.NullPool,
+            )
+            with connectable.connect() as connection:
+                context.configure(connection=connection)
+                with context.begin_transaction():
+                    context.run_migrations()
+
+        if context.is_offline_mode():
+            run_migrations_offline()
+        else:
+            run_migrations_online()
+    """)
+    )
+    ini = tmp_path / "alembic.ini"
+    ini.write_text(
+        _tw.dedent(f"""
+        [alembic]
+        script_location = {tmp_path}
+        sqlalchemy.url = sqlite:///{db}
+        version_locations = {versions}
+    """)
+    )
+    (versions / "001_create_users.py").write_text(
+        _tw.dedent("""
+        revision = '001'
+        down_revision = None
+        branch_labels = None
+        depends_on = None
+        import sqlalchemy as sa
+        from alembic import op
+        def upgrade():
+            op.create_table('users',
+                sa.Column('id', sa.Integer, primary_key=True),
+                sa.Column('name', sa.String(64), nullable=False),
+            )
+        def downgrade():
+            op.drop_table('users')
+    """)
+    )
+    return str(ini), str(versions), f"sqlite:///{db}"
+
+
+def test_drift_invalid_metadata_path_exits_1(tmp_path):
+    """mrt drift with a bad metadata path exits 1 with an error message."""
+    ini, _, db_url = _alembic_env(tmp_path)
+    result = runner.invoke(
+        app,
+        ["drift", "nonexistent.module:Base", "--config", ini, "--db-url", db_url],
+    )
+    assert result.exit_code == 1
+    assert "Error" in result.output
+
+
+def test_drift_missing_alembic_ini_exits_1(tmp_path):
+    """mrt drift exits 1 when alembic.ini does not exist."""
+    import sys
+    import types
+    import sqlalchemy as sa
+
+    # Register a valid metadata module so metadata loading succeeds,
+    # then the missing alembic.ini triggers the expected error.
+    mod = types.ModuleType("_test_drift_meta_noini")
+    base = sa.MetaData()
+    mod.Base = base
+    sys.modules["_test_drift_meta_noini"] = mod
+    try:
+        result = runner.invoke(
+            app,
+            ["drift", "_test_drift_meta_noini:Base", "--config", str(tmp_path / "nope.ini")],
+        )
+    finally:
+        sys.modules.pop("_test_drift_meta_noini", None)
+
+    assert result.exit_code == 1
+    assert "not found" in result.output.lower() or "no such file" in result.output.lower()
+
+
+def test_drift_no_drift_exits_0(tmp_path):
+    """mrt drift exits 0 and prints success when models match the DB."""
+    import sqlalchemy as sa
+
+    ini, _, db_url = _alembic_env(tmp_path)
+
+    # Register the metadata in sys.modules so load_metadata can find it
+    import sys
+    import types
+
+    mod = types.ModuleType("_mrt_test_models")
+    meta = sa.MetaData()
+    sa.Table(
+        "users",
+        meta,
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column("name", sa.String(64), nullable=False),
+    )
+    mod.Base = meta
+    sys.modules["_mrt_test_models"] = mod
+
+    try:
+        result = runner.invoke(
+            app,
+            ["drift", "_mrt_test_models:Base", "--config", ini, "--db-url", db_url],
+        )
+    finally:
+        sys.modules.pop("_mrt_test_models", None)
+
+    assert result.exit_code == 0
+    assert "No schema drift" in result.output
+
+
+def test_drift_with_drift_exits_1(tmp_path):
+    """mrt drift exits 1 and lists differences when the model has an extra column."""
+    import sqlalchemy as sa
+    import sys
+    import types
+
+    ini, _, db_url = _alembic_env(tmp_path)
+
+    mod = types.ModuleType("_mrt_test_drift_models")
+    meta = sa.MetaData()
+    sa.Table(
+        "users",
+        meta,
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column("name", sa.String(64), nullable=False),
+        sa.Column("email", sa.String(255)),  # extra — not in DB
+    )
+    mod.Base = meta
+    sys.modules["_mrt_test_drift_models"] = mod
+
+    try:
+        result = runner.invoke(
+            app,
+            ["drift", "_mrt_test_drift_models:Base", "--config", ini, "--db-url", db_url],
+        )
+    finally:
+        sys.modules.pop("_mrt_test_drift_models", None)
+
+    assert result.exit_code == 1
+    assert "difference" in result.output.lower()