test: coverage improvements, # mrt: ignore, Django dynamic verification fix (#21)

* test: cover _is_auto_pk SERIAL/nextval paths and json exit-1 on warnings

* feat: add # mrt: ignore annotation + fix Django dynamic verification

- Add # mrt: ignore line-level suppression in analyze_migrations()
  and analyze_django_migrations(): warnings whose triggering line
  contains the annotation are silently dropped, same as # noqa

- Fix Django dynamic verification: switch test fixture from
  sqlite:///:memory: (NullPool) to a named temp file so Django's
  connection and SQLAlchemy's engine see the same database; without
  this SchemaSnapshot comparisons were always vacuously equal

- Add tests/django_bad_app/ with a RunSQL noop-reverse migration
  and implement test_django_verifier_noop_downgrade_fails to confirm
  schema drift is detected (Issue #17)

Author: croc100

pytest_mrt/adapters/django_detector.py

@@ -512,7 +512,15 @@ def analyze_django_migrations(migrations_dir: str) -> list[RiskWarning]:
                 )
             )
             continue
+        source_lines = path.read_text().splitlines()
         for check in _DJANGO_CHECKS:
-            warnings.extend(check(m))
+            for w in check(m):
+                if (
+                    w.line is not None
+                    and w.line <= len(source_lines)
+                    and "# mrt: ignore" in source_lines[w.line - 1]
+                ):
+                    continue
+                warnings.append(w)
 
     return warnings

pytest_mrt/core/detector.py

@@ -814,8 +814,16 @@ def analyze_migrations(versions_dir: str) -> list[RiskWarning]:
             )
             continue
         migrations.append(m)
+        source_lines = source.splitlines()
         for check in _PER_FILE_CHECKS:
-            warnings.extend(check(m))
+            for w in check(m):
+                if (
+                    w.line is not None
+                    and w.line <= len(source_lines)
+                    and "# mrt: ignore" in source_lines[w.line - 1]
+                ):
+                    continue
+                warnings.append(w)
 
     warnings.extend(_check_multiple_heads(migrations))
     warnings.extend(analyze_migration_graph(versions_dir))

tests/django_bad_app/__init__.py

@@ -0,0 +1,23 @@
+"""
+Migration with a no-op downgrade — used to verify that DjangoRollbackVerifier
+detects schema drift when the rollback path does nothing.
+
+Forward:  creates a table via raw SQL
+Backward: RunSQL.noop — the table is never dropped, leaving schema drift.
+"""
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = []
+
+    operations = [
+        migrations.RunSQL(
+            sql=(
+                "CREATE TABLE django_bad_app_leaked "
+                "(id INTEGER PRIMARY KEY, val TEXT NOT NULL DEFAULT '')"
+            ),
+            reverse_sql=migrations.RunSQL.noop,
+        ),
+    ]

tests/django_bad_app/migrations/0001_noop_downgrade.py

@@ -111,6 +111,23 @@ def test_check_json_safe_exits_0(tmp_path, versions_dir):
     assert result.exit_code == 0
 
 
+def test_check_json_warnings_only_exits_1(tmp_path, versions_dir):
+    """--format json with warnings but no errors must exit 1."""
+    (versions_dir / "001.py").write_text(textwrap.dedent("""
+        revision = '001'
+        down_revision = None
+        branch_labels = None
+        depends_on = None
+        from alembic import op
+        def upgrade():
+            op.create_index('ix_name', 'users', ['name'])
+        def downgrade():
+            op.drop_index('ix_name', table_name='users')
+    """))
+    result = runner.invoke(app, ["check", str(versions_dir), "--format", "json"])
+    assert result.exit_code == 1
+
+
 def test_check_strict_exits_1_on_warnings(tmp_path, versions_dir):
     # A migration with only warnings (not errors) exits 0 normally
     (versions_dir / "001.py").write_text(textwrap.dedent("""

tests/django_bad_app/migrations/__init__.py

@@ -636,3 +636,41 @@ def test_syntax_error_migration_reported(tmp_path):
     warnings = analyze_migrations(str(tmp_path))
     assert any(w.pattern == "Syntax error" for w in warnings)
     assert any(w.severity == "error" for w in warnings)
+
+
+def test_mrt_ignore_suppresses_warning(tmp_path):
+    """# mrt: ignore on the same line suppresses that warning."""
+    migration(tmp_path, "001.py", """
+        revision = '001'
+        down_revision = None
+        branch_labels = None
+        depends_on = None
+        from alembic import op
+        def upgrade():
+            op.drop_column('users', 'email')  # mrt: ignore
+        def downgrade():
+            op.add_column('users', 'email')
+    """)
+    warnings = analyze_migrations(str(tmp_path))
+    assert not any(w.pattern == "DROP COLUMN in upgrade" for w in warnings)
+
+
+def test_mrt_ignore_only_suppresses_annotated_line(tmp_path):
+    """# mrt: ignore on one line does not suppress warnings on other lines."""
+    migration(tmp_path, "001.py", """
+        revision = '001'
+        down_revision = None
+        branch_labels = None
+        depends_on = None
+        from alembic import op
+        def upgrade():
+            op.drop_column('users', 'email')  # mrt: ignore
+            op.drop_column('users', 'phone')
+        def downgrade():
+            op.add_column('users', 'email')
+            op.add_column('users', 'phone')
+    """)
+    warnings = analyze_migrations(str(tmp_path))
+    drop_warnings = [w for w in warnings if w.pattern == "DROP COLUMN in upgrade"]
+    assert len(drop_warnings) == 1
+    assert "phone" in drop_warnings[0].message

tests/test_cli.py

@@ -384,3 +384,49 @@ class Migration(migrations.Migration):
     """)
     warnings = analyze_django_migrations(str(tmp_path))
     assert any(w.pattern == "Missing atomic=False" for w in warnings)
+
+
+def test_mrt_ignore_suppresses_django_warning(tmp_path):
+    """# mrt: ignore on the same line suppresses that Django migration warning."""
+    app_dir = tmp_path / "myapp" / "migrations"
+    app_dir.mkdir(parents=True, exist_ok=True)
+    (app_dir / "0002_remove.py").write_text(textwrap.dedent("""
+        from django.db import migrations
+
+        class Migration(migrations.Migration):
+            dependencies = [('myapp', '0001_initial')]
+            operations = [
+                migrations.RemoveField(  # mrt: ignore
+                    model_name='user',
+                    name='phone',
+                ),
+            ]
+    """).lstrip())
+    warnings = analyze_django_migrations(str(tmp_path))
+    assert not any(w.pattern == "RemoveField" for w in warnings)
+
+
+def test_mrt_ignore_only_suppresses_annotated_django_line(tmp_path):
+    """# mrt: ignore on one operation does not suppress others."""
+    app_dir = tmp_path / "myapp" / "migrations"
+    app_dir.mkdir(parents=True, exist_ok=True)
+    (app_dir / "0002_multi.py").write_text(textwrap.dedent("""
+        from django.db import migrations
+
+        class Migration(migrations.Migration):
+            dependencies = [('myapp', '0001_initial')]
+            operations = [
+                migrations.RemoveField(  # mrt: ignore
+                    model_name='user',
+                    name='phone',
+                ),
+                migrations.RemoveField(
+                    model_name='user',
+                    name='email',
+                ),
+            ]
+    """).lstrip())
+    warnings = analyze_django_migrations(str(tmp_path))
+    remove_warnings = [w for w in warnings if w.pattern == "RemoveField"]
+    assert len(remove_warnings) == 1
+    assert "email" in remove_warnings[0].message

tests/test_detector.py

@@ -2,7 +2,13 @@
 Django dynamic rollback integration tests.
 
 Tests DjangoMigrationRunner and DjangoRollbackVerifier using an in-process
-Django configuration with SQLite — no external database required.
+Django configuration with a file-backed SQLite database — no external database
+required, but Django must be installed.
+
+Uses a named SQLite file (via tmp_path_factory) so that both Django's own
+connection and SQLAlchemy's engine see the same database.  The :memory: URI
+with NullPool would give each engine.connect() call a fresh, empty database,
+making SchemaSnapshot comparisons meaningless.
 
 Skipped automatically when Django is not installed.
 """
@@ -20,26 +26,24 @@
     "django.contrib.contenttypes",
     "django.contrib.auth",
     "tests.django_app",
+    "tests.django_bad_app",
 ]
 
-DB_URL = "sqlite:///:memory:"
+
+# ── fixtures ──────────────────────────────────────────────────────────────────
 
 
-def _make_runner():
+@pytest.fixture(scope="module")
+def django_runner(tmp_path_factory):
     from pytest_mrt.adapters.django_runner import DjangoMigrationRunner
 
-    return DjangoMigrationRunner(
-        db_url=DB_URL,
+    db_path = tmp_path_factory.mktemp("django_db") / "test.db"
+    db_url = f"sqlite:///{db_path}"
+
+    runner = DjangoMigrationRunner(
+        db_url=db_url,
         installed_apps=INSTALLED_APPS,
     )
-
-
-# ── fixtures ──────────────────────────────────────────────────────────────────
-
-
-@pytest.fixture(scope="module")
-def django_runner():
-    runner = _make_runner()
     yield runner
     runner.dispose()
 
@@ -55,8 +59,8 @@ def test_django_runner_get_migrations(django_runner):
 
 
 def test_django_runner_upgrade_downgrade(django_runner):
-    """upgrade() then downgrade() leaves DB at original state."""
-    from pytest_mrt.core.schema import SchemaSnapshot
+    """upgrade() then downgrade() leaves DB schema unchanged."""
+    from pytest_mrt.core.schema import SchemaSnapshot, SchemaDiff
 
     migrations = django_runner.get_migrations(apps=["django_app"])
     assert migrations, "No migrations found in django_app"
@@ -69,8 +73,6 @@ def test_django_runner_upgrade_downgrade(django_runner):
     django_runner.downgrade(first.app_label, first.name)
 
     snap_after = SchemaSnapshot.capture(django_runner.engine)
-    from pytest_mrt.core.schema import SchemaDiff
-
     issues = SchemaDiff().verify_restored(snap_before, snap_after)
     assert issues == [], [i.message for i in issues]
 
@@ -87,11 +89,21 @@ def test_django_verifier_check_all_pass(django_runner):
     assert not failed, "\n".join(r.failure_summary() for r in failed)
 
 
-def test_django_verifier_noop_downgrade_fails(tmp_path):
-    """DjangoRollbackVerifier catches a no-op downgrade migration."""
-    pytest.skip(
-        "No-op downgrade detection requires a separate Django app fixture — "
-        "covered by static analysis (D-pattern checks)"
+def test_django_verifier_noop_downgrade_fails(django_runner):
+    """DjangoRollbackVerifier detects a migration whose downgrade is a no-op."""
+    from pytest_mrt.adapters.django_verifier import DjangoRollbackVerifier
+
+    verifier = DjangoRollbackVerifier(django_runner, timeout=30)
+    results = verifier.check_all(apps=["django_bad_app"])
+
+    assert results, "No results — django_bad_app migration not found"
+    # The first result must be a failure: schema drift (leaked table not dropped)
+    first = results[0]
+    assert not first.passed, (
+        "Expected noop downgrade to be detected as a failure, but it passed"
+    )
+    assert any("leaked" in f.lower() or "still exists" in f.lower() for f in first.failures), (
+        f"Expected 'still exists' schema error, got: {first.failures}"
     )
 
 
@@ -100,14 +112,14 @@ def test_mrt_config_django_mode():
     from pytest_mrt.config import MRTConfig
 
     cfg = MRTConfig(
-        db_url=DB_URL,
-        django_settings=None,  # not activating Django mode here
+        db_url="sqlite:///test.db",
+        django_settings=None,
     )
     assert cfg.django_settings is None
     assert cfg.django_apps is None
 
     cfg_django = MRTConfig(
-        db_url=DB_URL,
+        db_url="sqlite:///test.db",
         django_settings="myproject.settings_test",
         django_apps=["users"],
     )

tests/test_django_detector.py

@@ -263,6 +263,17 @@ def test_topological_order_self_reference():
     assert "nodes" in order
 
 
+def test_topological_order_mutual_fk_no_infinite_loop():
+    """Circular FK between two tables must not loop infinitely."""
+    tables = {
+        "a": TableInfo("a", fk_tables=["b"]),
+        "b": TableInfo("b", fk_tables=["a"]),
+    }
+    order = _topological_order(tables)
+    assert set(order) == {"a", "b"}
+    assert len(order) == 2
+
+
 # ── SmartSeeder ───────────────────────────────────────────────────────
 
 def test_seeder_seed_table_basic(engine):
@@ -421,3 +432,55 @@ def test_seeder_skips_nullable_columns(engine):
     assert len(rows) == 3
     # body is nullable, so it should be NULL in seeded rows
     assert all(row[1] is None for row in rows)
+
+
+# ── SmartSeeder._is_auto_pk ───────────────────────────────────────────
+
+def test_is_auto_pk_serial_type(engine):
+    """SERIAL type string is recognised as auto-generated PK."""
+    seeder = SmartSeeder(engine)
+    col = ColumnInfo(name="id", type_str="SERIAL", nullable=False, primary_key=True)
+    assert seeder._is_auto_pk(col) is True
+
+
+def test_is_auto_pk_auto_increment_type(engine):
+    """AUTO_INCREMENT (MySQL) is recognised as auto-generated PK."""
+    seeder = SmartSeeder(engine)
+    col = ColumnInfo(name="id", type_str="INT AUTO_INCREMENT", nullable=False, primary_key=True)
+    assert seeder._is_auto_pk(col) is True
+
+
+def test_is_auto_pk_nextval_default(engine):
+    """Sequence-backed PK detected via nextval() in the column default."""
+    seeder = SmartSeeder(engine)
+    col = ColumnInfo(name="id", type_str="INTEGER", nullable=False, primary_key=True,
+                     default="nextval('items_id_seq'::regclass)")
+    assert seeder._is_auto_pk(col) is True
+
+
+def test_is_auto_pk_plain_integer_is_not_auto(engine):
+    """Plain INTEGER PK with no default is not auto-generated."""
+    seeder = SmartSeeder(engine)
+    col = ColumnInfo(name="id", type_str="INTEGER", nullable=False, primary_key=True)
+    assert seeder._is_auto_pk(col) is False
+
+
+def test_seed_table_skips_serial_pk_row(engine):
+    """seed_table inserts nothing when the only non-nullable column is a SERIAL PK."""
+    with engine.begin() as conn:
+        conn.execute(text("CREATE TABLE only_serial (id INTEGER NOT NULL PRIMARY KEY)"))
+
+    seeder = SmartSeeder(engine)
+    # Construct TableInfo manually so type_str is SERIAL (as PG would reflect it)
+    table = TableInfo(
+        name="only_serial",
+        columns={
+            "id": ColumnInfo(name="id", type_str="SERIAL", nullable=False, primary_key=True),
+        },
+        pk_cols=["id"],
+    )
+    seeder.seed_table(table)
+
+    with engine.connect() as conn:
+        count = conn.execute(text("SELECT COUNT(*) FROM only_serial")).scalar()
+    assert count == 0