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

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()