Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 48 additions & 0 deletions pytest_mrt/adapters/django_fixer.py
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,9 @@ class DjangoFixSuggestion:
patched_source: str
confidence: str
warning: str | None = None
# Operations detected by mrt check but not auto-fixable (e.g. AddField NOT NULL,
# AlterField). Non-empty means the migration has issues mrt fix cannot resolve.
unsupported_ops: list[str] | None = None

@property
def issue(self) -> str:
Expand Down Expand Up @@ -197,6 +200,25 @@ def _has_kwarg(op: ast.Call, name: str) -> bool:
return any(kw.arg == name for kw in op.keywords)


def _add_field_is_problematic(op: ast.Call) -> bool:
"""Return True if AddField is NOT NULL with no default (MRT411 condition)."""
for kw in op.keywords:
if kw.arg == "field" and isinstance(kw.value, ast.Call):
field_call = kw.value
# null=True means it's safe
for fkw in field_call.keywords:
if fkw.arg == "null":
if isinstance(fkw.value, ast.Constant) and fkw.value.value is True:
return False
# has default= or server_default= -> safe
field_kwarg_names = {fkw.arg for fkw in field_call.keywords}
if "default" in field_kwarg_names or "server_default" in field_kwarg_names:
return False
# No null=True and no default -> problematic
return True
return False


def _has_positional_arg(op: ast.Call, index: int) -> bool:
return len(op.args) > index

Expand Down Expand Up @@ -536,7 +558,15 @@ def generate_django_fix(migration_path: str) -> DjangoFixSuggestion | None:
migration_name = f"{app_label}.{path.stem}"
migration_stem = path.stem

# Operations that mrt check may flag but mrt fix cannot auto-resolve.
# Only truly irreversible variants are listed — e.g. AddField(null=True) is
# fine, but AddField NOT NULL without a default is flagged by mrt check and
# cannot be auto-fixed. We detect these by name only (the detector handles
# the nuance); here we just surface them to the user.
_UNSUPPORTED_OPS = {"AlterField", "RenameField", "RenameModel"}

patches: list[DjangoOpPatch] = []
unsupported: list[str] = []

for op in ops_list:
if not isinstance(op, ast.Call):
Expand All @@ -552,11 +582,29 @@ def generate_django_fix(migration_path: str) -> DjangoFixSuggestion | None:
patch = _patch_remove_field(source, op, app_label, migration_stem)
elif name == "DeleteModel":
patch = _patch_delete_model(source, op, app_label, migration_stem)
elif name == "AddField":
# Only flag AddField that is NOT NULL with no default — same condition
# that mrt check (MRT411) flags. null=True or a default= means it's safe.
if _add_field_is_problematic(op):
unsupported.append("AddField NOT NULL without default")
elif name in _UNSUPPORTED_OPS:
unsupported.append(name)

if patch is not None:
patches.append(patch)

if not patches:
# Return a suggestion with no patches but unsupported ops listed,
# so the caller can show a meaningful message instead of "no fix needed".
if unsupported:
return DjangoFixSuggestion(
file=path.name,
migration_name=migration_name,
patches=[],
patched_source=source,
confidence="none",
unsupported_ops=unsupported,
)
return None

patched_source = _apply_patches_to_source(source, patches)
Expand Down
25 changes: 24 additions & 1 deletion pytest_mrt/commands/check.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,30 @@ def check(
is_django = any(is_django_migration(p) for p in sample_files)

if since:
console.print(f"[dim]--since {since}: checking only migrations after this point[/dim]")
# Validate that --since actually matches something before running analysis.
if is_django:
from ..adapters.django_detector import _django_migrations_since

since_set = _django_migrations_since(versions_dir, since)
else:
from ..core.detector import _revisions_since

since_set = _revisions_since(versions_dir, since)

if not since_set:
console.print(
f"[yellow]Warning: --since {since} matched no migrations. "
"Check the revision ID and try again.[/yellow]"
)
raise typer.Exit(1)

console.print(
f"[dim]--since {since}: checking {len(since_set)} migration(s) after this point[/dim]"
)
console.print(
"[dim]Note: graph checks (orphan, data-hole detection) skipped — "
"run without --since for full analysis.[/dim]"
)

if is_django:
console.print("[dim]Detected: Django migrations[/dim]")
Expand Down
12 changes: 12 additions & 0 deletions pytest_mrt/commands/fix.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,18 @@ def _fix_django(migration_file: str, apply: bool) -> None:
)
raise typer.Exit(0)

if fix_suggestion.unsupported_ops and not fix_suggestion.patches:
ops = ", ".join(sorted(set(fix_suggestion.unsupported_ops)))
console.print(
f"[yellow]mrt fix cannot auto-fix this migration.[/yellow]\n"
f"Operation(s) found: [bold]{ops}[/bold]\n\n"
"These operations must be fixed manually:\n"
" - AddField / AlterField: add server_default or handle nullable migration\n"
" - RenameField / RenameModel: verify old_name is correct and reversible\n\n"
"Run [bold]mrt check[/bold] to see the specific warnings."
)
raise typer.Exit(1)

console.print()
console.print(
f"[bold]{fix_suggestion.file}[/bold] [dim]{fix_suggestion.migration_name}[/dim]"
Expand Down
50 changes: 50 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,56 @@ def test_check_file_instead_of_dir_exits_1(tmp_path):
assert "not a directory" in result.output


# ── mrt check --since ────────────────────────────


def test_check_since_unknown_revision_exits_1(tmp_path, versions_dir):
"""--since with an unknown revision exits 1 with a clear warning."""
_safe_migration(versions_dir)
result = runner.invoke(app, ["check", str(versions_dir), "--since", "deadbeef"])
assert result.exit_code == 1
assert "matched no migrations" in result.output


def test_check_since_valid_revision_shows_note(tmp_path, versions_dir):
"""--since with a valid revision prints the graph-checks-skipped note."""
# migration 001 (down_revision=None), migration 002 (down_revision=001)
_safe_migration(versions_dir, name="001_safe.py")
(versions_dir / "002_after.py").write_text(
textwrap.dedent("""
revision = '002'
down_revision = '001'
from alembic import op
def upgrade():
op.create_table('posts', )
def downgrade():
op.drop_table('posts')
""")
)
result = runner.invoke(app, ["check", str(versions_dir), "--since", "001"])
assert result.exit_code == 0
assert "graph checks" in result.output
assert "skipped" in result.output


def test_check_since_valid_revision_shows_count(tmp_path, versions_dir):
"""--since prints how many migrations are being checked."""
_safe_migration(versions_dir, name="001_safe.py")
(versions_dir / "002_after.py").write_text(
textwrap.dedent("""
revision = '002'
down_revision = '001'
from alembic import op
def upgrade():
pass
def downgrade():
pass
""")
)
result = runner.invoke(app, ["check", str(versions_dir), "--since", "001"])
assert "1 migration" in result.output


# ── mrt init Django detection ────────────────────


Expand Down
20 changes: 20 additions & 0 deletions tests/test_django_fixer.py
Original file line number Diff line number Diff line change
Expand Up @@ -400,6 +400,7 @@ def test_issue_multiple(tmp_path):


def test_fix_command_routes_to_django(tmp_path):
"""mrt fix on a Django migration with AddField NOT NULL shows a clear error."""
from typer.testing import CliRunner

from pytest_mrt.cli import app
Expand All @@ -412,6 +413,25 @@ def test_fix_command_routes_to_django(tmp_path):
)
runner = CliRunner()
result = runner.invoke(app, ["fix", p])
assert result.exit_code == 1
assert "cannot auto-fix" in result.output
assert "AddField" in result.output


def test_fix_command_routes_to_django_nullable(tmp_path):
"""mrt fix on AddField(null=True) reports no fix needed — it is reversible."""
from typer.testing import CliRunner

from pytest_mrt.cli import app

p = _write(
tmp_path,
HEADER
+ " migrations.AddField(model_name='u', name='x', field=models.TextField(null=True)),\n"
+ FOOTER,
)
runner = CliRunner()
result = runner.invoke(app, ["fix", p])
assert result.exit_code == 0
assert "No fix needed" in result.output

Expand Down
Loading