diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 5b84b183c..da8b9502a 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -28,3 +28,12 @@ repos: rev: 7.0.0 hooks: - id: flake8 + + - repo: local + hooks: + - id: lint-django-migrations + name: Lint Django migrations + entry: python manage.py lint_migrations + language: system + pass_filenames: false + files: migrations/.*\.py$ diff --git a/api/management/commands/lint_migrations.py b/api/management/commands/lint_migrations.py new file mode 100644 index 000000000..fda95735c --- /dev/null +++ b/api/management/commands/lint_migrations.py @@ -0,0 +1,39 @@ +from pathlib import Path + +from django.conf import settings +from django.core.management.base import BaseCommand, CommandError + +from api.migration_lint import lint_migrations + + +class Command(BaseCommand): + help = "Lint Django migration files for naming, duplicate numbers, and dependency issues." + + def add_arguments(self, parser): + parser.add_argument( + "--warnings-as-errors", + action="store_true", + help="Treat warnings as errors (useful in CI).", + ) + + def handle(self, *args, **options): + base_dir = Path(settings.BASE_DIR) + result = lint_migrations(base_dir) + + for issue in result.issues: + line = f"[{issue.level.upper()}] {issue.app}/{issue.path}: {issue.message}" + if issue.level == "error": + self.stderr.write(line) + else: + self.stdout.write(line) + + if not result.issues: + self.stdout.write(self.style.SUCCESS("No migration lint issues found.")) + return + + has_errors = not result.ok + has_warnings = any(issue.level == "warning" for issue in result.issues) + if has_errors or (options["warnings_as_errors"] and has_warnings): + raise CommandError("Migration lint failed.") + + self.stdout.write(self.style.WARNING("Migration lint completed with warnings only.")) diff --git a/api/migration_lint.py b/api/migration_lint.py new file mode 100644 index 000000000..549883472 --- /dev/null +++ b/api/migration_lint.py @@ -0,0 +1,179 @@ +"""Utilities for validating Django migration files.""" + +from __future__ import annotations + +import ast +import re +from dataclasses import dataclass, field +from pathlib import Path + +MIGRATION_FILENAME_RE = re.compile(r"^(?P\d{4})_(?P.+)\.py$") +MERGE_SLUG_RE = re.compile(r"^merge_") + + +@dataclass +class MigrationLintIssue: + level: str # error | warning + app: str + path: str + message: str + + +@dataclass +class MigrationLintResult: + issues: list[MigrationLintIssue] = field(default_factory=list) + + @property + def ok(self) -> bool: + return not any(issue.level == "error" for issue in self.issues) + + +def _ast_value(node: ast.AST): + if isinstance(node, ast.Constant): + return node.value + if isinstance(node, ast.Tuple): + return tuple(_ast_value(elt) for elt in node.elts) + if isinstance(node, ast.List): + return [_ast_value(elt) for elt in node.elts] + if isinstance(node, ast.Call): + # migrations.swappable_dependency(...) — resolved by Django at runtime. + return None + raise ValueError(f"Unsupported AST node in migration file: {type(node).__name__}") + + +def _parse_dependencies(path: Path) -> list[tuple[str, str]]: + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in tree.body: + if isinstance(node, ast.ClassDef) and node.name == "Migration": + for item in node.body: + if isinstance(item, ast.Assign): + for target in item.targets: + if isinstance(target, ast.Name) and target.id == "dependencies": + return _ast_value(item.value) + return [] + + +def _discover_migration_apps(base_dir: Path) -> list[tuple[str, Path]]: + apps: list[tuple[str, Path]] = [] + for migrations_dir in sorted(base_dir.glob("*/migrations")): + if not migrations_dir.is_dir(): + continue + app_label = migrations_dir.parent.name + if app_label.startswith("."): + continue + apps.append((app_label, migrations_dir)) + return apps + + +def lint_migrations(base_dir: Path) -> MigrationLintResult: + result = MigrationLintResult() + known: dict[tuple[str, str], Path] = {} + by_app_number: dict[tuple[str, str], list[str]] = {} + parsed: dict[tuple[str, str], list[tuple[str, str]]] = {} + + for app_label, migrations_dir in _discover_migration_apps(base_dir): + for path in sorted(migrations_dir.glob("*.py")): + if path.name == "__init__.py": + continue + + match = MIGRATION_FILENAME_RE.match(path.name) + if not match: + result.issues.append( + MigrationLintIssue( + level="error", + app=app_label, + path=str(path.relative_to(base_dir)), + message=( + "Invalid migration filename; expected NNNN_description.py " + f"(got {path.name})" + ), + ) + ) + continue + + number = match.group("number") + slug = match.group("slug") + migration_name = path.stem + known[(app_label, migration_name)] = path + by_app_number.setdefault((app_label, number), []).append(migration_name) + + try: + raw_deps = _parse_dependencies(path) + parsed[(app_label, migration_name)] = [ + dep for dep in raw_deps if isinstance(dep, tuple) and len(dep) == 2 + ] + except (SyntaxError, ValueError) as exc: + result.issues.append( + MigrationLintIssue( + level="error", + app=app_label, + path=str(path.relative_to(base_dir)), + message=f"Could not parse dependencies: {exc}", + ) + ) + + if slug.startswith("_"): + result.issues.append( + MigrationLintIssue( + level="warning", + app=app_label, + path=str(path.relative_to(base_dir)), + message="Migration slug should not start with an underscore", + ) + ) + + for (app_label, number), names in by_app_number.items(): + if len(names) <= 1: + continue + + branch_names = [ + name + for name in names + if not MERGE_SLUG_RE.match(name.split("_", 1)[1] if "_" in name else "") + ] + if len(branch_names) <= 1: + continue + + merged = False + for (merge_app, merge_name), deps in parsed.items(): + if merge_app != app_label: + continue + slug = merge_name.split("_", 1)[1] if "_" in merge_name else merge_name + if not MERGE_SLUG_RE.match(slug): + continue + dep_names = {dep_name for dep_app, dep_name in deps if dep_app == app_label} + if set(branch_names).issubset(dep_names): + merged = True + break + + if not merged: + result.issues.append( + MigrationLintIssue( + level="warning", + app=app_label, + path=app_label, + message=( + f"Duplicate migration number {number} in app '{app_label}': " + f"{', '.join(sorted(branch_names))}. Add a merge migration if this is intentional." + ), + ) + ) + + app_labels = {app for app, _ in known} + + for (app_label, migration_name), dependencies in parsed.items(): + rel_path = known[(app_label, migration_name)].relative_to(base_dir) + for dep_app, dep_name in dependencies: + if dep_app not in app_labels: + continue + if (dep_app, dep_name) not in known: + result.issues.append( + MigrationLintIssue( + level="error", + app=app_label, + path=str(rel_path), + message=f"Missing dependency target: ({dep_app!r}, {dep_name!r})", + ) + ) + + return result diff --git a/api/tests/__init__.py b/api/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/api/tests/test_migration_lint.py b/api/tests/test_migration_lint.py new file mode 100644 index 000000000..6b3f08c79 --- /dev/null +++ b/api/tests/test_migration_lint.py @@ -0,0 +1,52 @@ +import tempfile +import unittest +from pathlib import Path + +from api.migration_lint import lint_migrations + +REPO_ROOT = Path(__file__).resolve().parents[2] + + +class MigrationLintTests(unittest.TestCase): + def test_existing_repo_migrations_have_valid_dependencies(self): + result = lint_migrations(REPO_ROOT) + errors = [issue for issue in result.issues if issue.level == "error"] + self.assertEqual(errors, []) + + def test_flags_invalid_filename(self): + with tempfile.TemporaryDirectory() as tempdir: + base = Path(tempdir) + migrations_dir = base / "sampleapp" / "migrations" + migrations_dir.mkdir(parents=True) + (migrations_dir / "__init__.py").write_text("") + (migrations_dir / "bad-name.py").write_text( + "from django.db import migrations\n\n" + "class Migration(migrations.Migration):\n" + " dependencies = []\n" + " operations = []\n" + ) + + result = lint_migrations(base) + self.assertFalse(result.ok) + self.assertTrue(any("Invalid migration filename" in issue.message for issue in result.issues)) + + def test_flags_missing_dependency(self): + with tempfile.TemporaryDirectory() as tempdir: + base = Path(tempdir) + migrations_dir = base / "sampleapp" / "migrations" + migrations_dir.mkdir(parents=True) + (migrations_dir / "__init__.py").write_text("") + (migrations_dir / "0001_initial.py").write_text( + "from django.db import migrations\n\n" + "class Migration(migrations.Migration):\n" + " dependencies = [('sampleapp', '9999_missing')]\n" + " operations = []\n" + ) + + result = lint_migrations(base) + self.assertFalse(result.ok) + self.assertTrue(any("Missing dependency target" in issue.message for issue in result.issues)) + + +if __name__ == "__main__": + unittest.main()