diff --git a/pytest_mrt/adapters/django_fixer.py b/pytest_mrt/adapters/django_fixer.py index 7dde345..b8c0d2c 100644 --- a/pytest_mrt/adapters/django_fixer.py +++ b/pytest_mrt/adapters/django_fixer.py @@ -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: @@ -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 @@ -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): @@ -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) diff --git a/pytest_mrt/commands/check.py b/pytest_mrt/commands/check.py index aa1d3f2..d7c3b37 100644 --- a/pytest_mrt/commands/check.py +++ b/pytest_mrt/commands/check.py @@ -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]") diff --git a/pytest_mrt/commands/fix.py b/pytest_mrt/commands/fix.py index cfee47e..165ec0f 100644 --- a/pytest_mrt/commands/fix.py +++ b/pytest_mrt/commands/fix.py @@ -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]" diff --git a/tests/test_cli.py b/tests/test_cli.py index 85cf50f..784a589 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -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 ──────────────────── diff --git a/tests/test_django_fixer.py b/tests/test_django_fixer.py index a7619e5..6c4678b 100644 --- a/tests/test_django_fixer.py +++ b/tests/test_django_fixer.py @@ -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 @@ -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