fix: Django fixer E2E tests, --since docs, mrt explain file check order + tests (#44)

Author: croc100

README.md

@@ -174,8 +174,11 @@ Add to `.pre-commit-config.yaml` to run `mrt check` automatically before every p
 Check only migrations added since a given revision. Keeps CI fast on large codebases:
 
 ```bash
-mrt check migrations/versions/ --since main
-mrt check myapp/migrations/ --since v1.2.0
+# Alembic — pass a revision ID
+mrt check migrations/versions/ --since a1b2c3d4
+
+# Django — pass app_label.migration_name
+mrt check myapp/migrations/ --since myapp.0010_add_email
 ```
 
 ## CI/CD integration

docs/cli.md

@@ -61,15 +61,21 @@ mrt check migrations/versions/
 
 #### `--since` — incremental scanning
 
-In large codebases, scanning the full migration history on every PR is wasteful. `--since` limits the scan to migrations whose files were added after the given git ref:
+In large codebases, scanning the full migration history on every PR is wasteful. `--since` limits the scan to migrations that come *after* the given revision in the migration dependency chain.
 
+**Alembic** — pass a revision ID:
 ```bash
-mrt check migrations/versions/ --since main
-mrt check myapp/migrations/ --since v1.2.0
-mrt check myapp/migrations/ --since HEAD~5
+mrt check migrations/versions/ --since a1b2c3d4
 ```
+Only revisions whose `down_revision` ancestry includes `a1b2c3d4` are checked.
 
-This is the recommended setup for CI: check only the migrations that changed in the current PR branch.
+**Django** — pass `app_label.migration_name`:
+```bash
+mrt check myapp/migrations/ --since myapp.0010_add_email
+```
+Only migrations that depend on `0010_add_email` (directly or transitively) are checked.
+
+This is the recommended CI pattern: pass the last migration on the base branch so only the PR's new migrations are scanned.
 
 ### Exit codes
 

pytest_mrt/commands/output.py

@@ -38,6 +38,11 @@ def explain(
     Requires: pip install pytest-mrt[ai]
     Requires: ANTHROPIC_API_KEY environment variable
     """
+    path = Path(migration_file)
+    if not path.exists():
+        console.print(f"[red]File not found: {migration_file}[/red]")
+        raise typer.Exit(1)
+
     try:
         import anthropic
     except ImportError:
@@ -50,11 +55,6 @@ def explain(
         )
         raise typer.Exit(1)
 
-    path = Path(migration_file)
-    if not path.exists():
-        console.print(f"[red]File not found: {migration_file}[/red]")
-        raise typer.Exit(1)
-
     source = path.read_text()
 
     console.print(f"[dim]Analyzing {path.name}...[/dim]")

tests/django_fixer_app/__init__.py

@@ -0,0 +1,18 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    initial = True
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="Contact",
+            fields=[
+                ("id", models.AutoField(primary_key=True, serialize=False)),
+                ("name", models.CharField(max_length=128)),
+                ("phone", models.CharField(blank=True, max_length=32, null=True)),
+            ],
+            options={"app_label": "django_fixer_app"},
+        ),
+    ]

tests/django_fixer_app/migrations/0001_initial.py

@@ -0,0 +1,140 @@
+# Reference migration: matches the pattern that `mrt fix` generates for RemoveField.
+# Used by test_django_fixer_e2e.py to verify that the generated code pattern
+# actually executes correctly against a real database.
+from django.db import migrations
+
+_MRT_TABLE = "_mrt_backups"
+_MRT_CHUNK = 500
+
+
+def __mrt_enc(v):
+    import base64
+    from datetime import date, datetime, time
+    from decimal import Decimal
+    from uuid import UUID
+
+    if v is None:
+        return None
+    if isinstance(v, bool):
+        return v
+    if isinstance(v, int):
+        return v
+    if isinstance(v, float):
+        return v
+    if isinstance(v, Decimal):
+        return "D:" + str(v)
+    if isinstance(v, datetime):
+        if v.tzinfo is not None:
+            return "DTs:" + str(v.timestamp())
+        return "DT:" + v.isoformat()
+    if isinstance(v, date):
+        return "d:" + v.isoformat()
+    if isinstance(v, time):
+        return "t:" + v.isoformat()
+    if isinstance(v, UUID):
+        return "U:" + str(v)
+    if isinstance(v, (bytes, bytearray, memoryview)):
+        return "B:" + base64.b64encode(bytes(v)).decode()
+    s = str(v)
+    for prefix in ("D:", "DT:", "DTs:", "d:", "t:", "U:", "B:", "S:"):
+        if s.startswith(prefix):
+            return "S:" + s
+    return s
+
+
+def __mrt_dec(v):
+    import base64
+    from datetime import date, datetime, time
+    from decimal import Decimal
+    from uuid import UUID
+
+    if not isinstance(v, str):
+        return v
+    if v.startswith("D:"):
+        return Decimal(v[2:])
+    if v.startswith("DTs:"):
+        return datetime.fromtimestamp(float(v[4:]))
+    if v.startswith("DT:"):
+        return datetime.fromisoformat(v[3:])
+    if v.startswith("d:"):
+        return date.fromisoformat(v[2:])
+    if v.startswith("t:"):
+        return time.fromisoformat(v[2:])
+    if v.startswith("U:"):
+        return UUID(v[2:])
+    if v.startswith("B:"):
+        return base64.b64decode(v[2:])
+    if v.startswith("S:"):
+        return v[2:]
+    return v
+
+
+_MRT_LABEL_contact_phone = "0002_remove_phone__contact_phone"
+
+
+def _backup_contact_phone(apps, schema_editor):
+    import json
+
+    from django.db import connection
+
+    Contact = apps.get_model("django_fixer_app", "Contact")
+    with connection.cursor() as cur:
+        cur.execute(
+            "CREATE TABLE IF NOT EXISTS " + _MRT_TABLE + " "
+            "(migration_label TEXT NOT NULL, object_id TEXT NOT NULL, payload TEXT)"
+        )
+        cur.execute(
+            "DELETE FROM " + _MRT_TABLE + " WHERE migration_label = %s",
+            [_MRT_LABEL_contact_phone],
+        )
+    last_pk = None
+    while True:
+        qs = Contact.objects.order_by("pk")
+        if last_pk is not None:
+            qs = qs.filter(pk__gt=last_pk)
+        batch = list(qs.values_list("pk", "phone")[:_MRT_CHUNK])
+        if not batch:
+            break
+        with connection.cursor() as cur:
+            for pk, val in batch:
+                cur.execute(
+                    "INSERT INTO " + _MRT_TABLE + " VALUES (%s, %s, %s)",
+                    [
+                        _MRT_LABEL_contact_phone,
+                        json.dumps(__mrt_enc(pk)),
+                        json.dumps(__mrt_enc(val)),
+                    ],
+                )
+        last_pk = batch[-1][0]
+
+
+def _restore_contact_phone(apps, schema_editor):
+    import json
+
+    from django.db import connection
+
+    Contact = apps.get_model("django_fixer_app", "Contact")
+    with connection.cursor() as cur:
+        cur.execute(
+            "SELECT object_id, payload FROM " + _MRT_TABLE + " WHERE migration_label = %s",
+            [_MRT_LABEL_contact_phone],
+        )
+        rows = cur.fetchall()
+    for pk_raw, val_raw in rows:
+        pk = __mrt_dec(json.loads(pk_raw))
+        val = __mrt_dec(json.loads(val_raw))
+        Contact.objects.filter(pk=pk).update(phone=val)
+    with connection.cursor() as cur:
+        cur.execute(
+            "DELETE FROM " + _MRT_TABLE + " WHERE migration_label = %s",
+            [_MRT_LABEL_contact_phone],
+        )
+
+
+class Migration(migrations.Migration):
+    dependencies = [("django_fixer_app", "0001_initial")]
+
+    operations = [
+        migrations.RunPython(_backup_contact_phone, _restore_contact_phone),
+        migrations.RemoveField(model_name="Contact", name="phone"),
+    ]

tests/django_fixer_app/migrations/0002_remove_phone.py

@@ -0,0 +1,9 @@
+from django.db import models
+
+
+class Contact(models.Model):
+    name = models.CharField(max_length=128)
+    phone = models.CharField(max_length=32, null=True, blank=True)
+
+    class Meta:
+        app_label = "django_fixer_app"

tests/django_fixer_app/migrations/__init__.py

@@ -1,4 +1,5 @@
 """Tests for MigrationAST helper methods in ast_analyzer.py."""
+
 from __future__ import annotations
 
 import ast
@@ -13,6 +14,7 @@ def _ast(source: str) -> MigrationAST:
 
 # ── module_var ────────────────────────────────────────────────────────
 
+
 def test_module_var_string():
     m = _ast("""
         revision = '001'
@@ -44,13 +46,15 @@ def downgrade(): pass
 
 # ── parse error ───────────────────────────────────────────────────────
 
+
 def test_parse_error_sets_flag():
     m = MigrationAST("def upgrade(: pass\n", "001", "001.py")
     assert m._parse_error is not None
 
 
 # ── is_noop ───────────────────────────────────────────────────────────
 
+
 def test_is_noop_with_pass():
     m = _ast("""
         revision = '001'
@@ -82,6 +86,7 @@ def upgrade(): pass
 
 # ── str_arg ───────────────────────────────────────────────────────────
 
+
 def test_str_arg_returns_string():
     tree = ast.parse("op.drop_table('users')")
     call = next(n for n in ast.walk(tree) if isinstance(n, ast.Call))
@@ -102,6 +107,7 @@ def test_str_arg_non_constant_returns_none():
 
 # ── has_kwarg ─────────────────────────────────────────────────────────
 
+
 def test_has_kwarg_true():
     tree = ast.parse("op.alter_column('t', 'c', nullable=False)")
     call = next(n for n in ast.walk(tree) if isinstance(n, ast.Call))
@@ -116,6 +122,7 @@ def test_has_kwarg_false():
 
 # ── kwarg_str ─────────────────────────────────────────────────────────
 
+
 def test_kwarg_str_returns_value():
     tree = ast.parse("op.create_index('ix', 'users', schema='public')")
     call = next(n for n in ast.walk(tree) if isinstance(n, ast.Call))
@@ -130,6 +137,7 @@ def test_kwarg_str_missing_returns_none():
 
 # ── kwarg_bool ────────────────────────────────────────────────────────
 
+
 def test_kwarg_bool_true():
     tree = ast.parse("op.alter_column('t', 'c', nullable=True)")
     call = next(n for n in ast.walk(tree) if isinstance(n, ast.Call))
@@ -156,6 +164,7 @@ def test_kwarg_bool_missing_returns_none():
 
 # ── sql_content ───────────────────────────────────────────────────────
 
+
 def test_sql_content_plain_string():
     tree = ast.parse("op.execute('SELECT 1')")
     call = next(n for n in ast.walk(tree) if isinstance(n, ast.Call))
@@ -177,20 +186,17 @@ def test_sql_content_no_args():
 
 # ── find_column_calls ─────────────────────────────────────────────────
 
+
 def test_find_column_calls_sa_column():
-    tree = ast.parse(
-        "op.create_table('t', sa.Column('id', sa.Integer, primary_key=True))"
-    )
+    tree = ast.parse("op.create_table('t', sa.Column('id', sa.Integer, primary_key=True))")
     calls = [n for n in ast.walk(tree) if isinstance(n, ast.Call)]
     outer = calls[0]
     cols = MigrationAST.find_column_calls(outer)
     assert len(cols) >= 1
 
 
 def test_find_column_calls_bare_column():
-    tree = ast.parse(
-        "op.create_table('t', Column('id', Integer, primary_key=True))"
-    )
+    tree = ast.parse("op.create_table('t', Column('id', Integer, primary_key=True))")
     calls = [n for n in ast.walk(tree) if isinstance(n, ast.Call)]
     outer = calls[0]
     cols = MigrationAST.find_column_calls(outer)
@@ -199,6 +205,7 @@ def test_find_column_calls_bare_column():
 
 # ── upgrade_methods / downgrade_methods ───────────────────────────────
 
+
 def test_upgrade_methods_returns_set():
     m = _ast("""
         revision = '001'
@@ -216,6 +223,7 @@ def downgrade():
 
 # ── nested function not counted ───────────────────────────────────────
 
+
 def test_nested_function_calls_not_attributed_to_upgrade(tmp_path):
     """Calls inside a nested def inside upgrade() should not appear in upgrade_calls."""
     m = _ast("""