Repository: 3YOURMIND/django-migration-linter
Pull request #123 (2020-11-16 21:14:19)

Data migration detection improvements
Fix https://github.com/3YOURMIND/django-migration-linter/issues/121

* Make data migration model import error less strict
* Add warning detection on RunPython call when model variable name is not the same as model class name
* Run checks on RunSQL migration operations

I'll add some more tests and feature on the RunSQL call


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 80c4869..629983a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,9 @@
 ## 2.5.0
 
 * Add Python 3.9 support
+* Make data migration model import error less strict (issue #121)
+* Add warning detection on RunPython call when model variable name is not the same as model class name
+* Run checks on RunSQL migration operations
 
 Switched from Travis CI to GitHub Actions for tests
 
diff --git a/django_migration_linter/migration_linter.py b/django_migration_linter/migration_linter.py
index 4e27826..9a3d006 100644
--- a/django_migration_linter/migration_linter.py
+++ b/django_migration_linter/migration_linter.py
@@ -10,7 +10,7 @@ from subprocess import Popen, PIPE
 from django.conf import settings
 from django.core.management import call_command
 from django.db import DEFAULT_DB_ALIAS, connections, ProgrammingError
-from django.db.migrations import RunPython
+from django.db.migrations import RunPython, RunSQL
 from enum import Enum, unique
 from six import PY2
 
@@ -392,12 +392,17 @@ class MigrationLinter(object):
         for operation in migration.operations:
             if isinstance(operation, RunPython):
                 op_errors, op_ignored, op_warnings = self.lint_runpython(operation)
-                if op_errors:
-                    errors += op_errors
-                if op_ignored:
-                    ignored += op_ignored
-                if op_warnings:
-                    warnings += op_warnings
+            elif isinstance(operation, RunSQL):
+                op_errors, op_ignored, op_warnings = self.lint_runsql(operation)
+            else:
+                op_errors, op_ignored, op_warnings = [], [], []
+
+            if op_errors:
+                errors += op_errors
+            if op_ignored:
+                ignored += op_ignored
+            if op_warnings:
+                warnings += op_warnings
 
         return errors, ignored, warnings
 
@@ -407,6 +412,7 @@ class MigrationLinter(object):
         ignored = []
         warning = []
 
+        # Detect warning on missing reverse operation
         if not runpython.reversible:
             issue = {
                 "code": "REVERSIBLE_DATA_MIGRATION",
@@ -419,11 +425,12 @@ class MigrationLinter(object):
             else:
                 warning.append(issue)
 
-        if PY2:
-            args_spec = inspect.getargspec(runpython.code)
-        else:
-            args_spec = inspect.getfullargspec(runpython.code)
-
+        # Detect warning for argument naming convention
+        args_spec = (
+            inspect.getargspec(runpython.code)
+            if PY2
+            else inspect.getfullargspec(runpython.code)
+        )
         if tuple(args_spec.args) != EXPECTED_DATA_MIGRATION_ARGS:
             issue = {
                 "code": "NAMING_CONVENTION_RUNPYTHON_ARGS",
@@ -439,7 +446,7 @@ class MigrationLinter(object):
 
         # Detect wrong model imports
         # Forward
-        issues = self.get_data_migration_model_import_issues(runpython.code)
+        issues = self.get_runpython_model_import_issues(runpython.code)
         for issue in issues:
             if issue["code"] in self.exclude_migration_tests:
                 ignored.append(issue)
@@ -448,28 +455,47 @@ class MigrationLinter(object):
 
         # Backward
         if runpython.reversible:
-            issues = self.get_data_migration_model_import_issues(runpython.reverse_code)
+            issues = self.get_runpython_model_import_issues(runpython.reverse_code)
             for issue in issues:
                 if issue and issue["code"] in self.exclude_migration_tests:
                     ignored.append(issue)
                 else:
                     error.append(issue)
+
+        # Detect warning if model variable name is not the same as model class
+        issues = self.get_runpython_model_variable_naming_issues(runpython.code)
+        for issue in issues:
+            if issue and issue["code"] in self.exclude_migration_tests:
+                ignored.append(issue)
+            else:
+                warning.append(issue)
+
+        if runpython.reversible:
+            issues = self.get_runpython_model_variable_naming_issues(
+                runpython.reverse_code
+            )
+            for issue in issues:
+                if issue and issue["code"] in self.exclude_migration_tests:
+                    ignored.append(issue)
+                else:
+                    warning.append(issue)
+
         return error, ignored, warning
 
     @staticmethod
-    def get_data_migration_model_import_issues(code):
+    def get_runpython_model_import_issues(code):
         model_object_regex = re.compile(r"[^a-zA-Z]?([a-zA-Z0-9]+?)\.objects")
 
         function_name = code.__name__
-        reverse_code_source = inspect.getsource(code)
+        source_code = inspect.getsource(code)
 
-        called_models = model_object_regex.findall(reverse_code_source)
+        called_models = model_object_regex.findall(source_code)
         issues = []
         for model in called_models:
             has_get_model_call = (
                 re.search(
-                    r"get_model\(\s*?.*?,.*?{}.+?\s*?\)".format(model),
-                    reverse_code_source,
+                    r"{}.*= +\w+\.get_model\(".format(model),
+                    source_code,
                 )
                 is not None
             )
@@ -485,3 +511,118 @@ class MigrationLinter(object):
                     }
                 )
         return issues
+
+    @staticmethod
+    def get_runpython_model_variable_naming_issues(code):
+        model_object_regex = re.compile(r"[^a-zA-Z]?([a-zA-Z0-9]+?)\.objects")
+
+        function_name = code.__name__
+        source_code = inspect.getsource(code)
+
+        called_models = model_object_regex.findall(source_code)
+        issues = []
+        for model in called_models:
+            has_same_model_name = (
+                re.search(
+                    r"{model}.*= +\w+?\.get_model\([^)]+?\.{model}.*?\)".format(
+                        model=model
+                    ),
+                    source_code,
+                    re.MULTILINE | re.DOTALL,
+                )
+                is not None
+                or re.search(
+                    r"{model}.*= +\w+?\.get_model\([^)]+?,[^)]*?{model}.*?\)".format(
+                        model=model
+                    ),
+                    source_code,
+                    re.MULTILINE | re.DOTALL,
+                )
+                is not None
+            )
+            if not has_same_model_name:
+                issues.append(
+                    {
+                        "code": "DATA_MIGRATION_MODEL_VARIABLE_NAME",
+                        "msg": (
+                            "'{}': Model variable name {} is different from the "
+                            "model class name that was found in the "
+                            "apps.get_model(...) call."
+                        ).format(function_name, model),
+                    }
+                )
+        return issues
+
+    def lint_runsql(self, runsql):
+        error = []
+        ignored = []
+        warning = []
+
+        # Detect warning on missing reverse operation
+        if not runsql.reversible:
+            issue = {
+                "code": "REVERSIBLE_RUNSQL_DATA_MIGRATION",
+                "msg": "RunSQL data migration is not reversible",
+            }
+            if issue["code"] in self.exclude_migration_tests:
+                ignored.append(issue)
+            else:
+                warning.append(issue)
+
+        # Put the SQL in our SQL analyser
+        if runsql.sql != RunSQL.noop:
+            sql_statements = []
+            if isinstance(runsql.sql, (list, tuple)):
+                for sql in runsql.sql:
+                    params = None
+                    if isinstance(sql, (list, tuple)):
+                        elements = len(sql)
+                        if elements == 2:
+                            sql, params = sql
+                        else:
+                            raise ValueError("Expected a 2-tuple but got %d" % elements)
+                        sql_statements.append(sql % params)
+                    else:
+                        sql_statements.append(sql)
+            else:
+                sql_statements.append(runsql.sql)
+
+            sql_errors, sql_ignored = analyse_sql_statements(
+                sql_statements,
+                settings.DATABASES[self.database]["ENGINE"],
+                self.exclude_migration_tests,
+            )
+            if sql_errors:
+                error += sql_errors
+            if sql_ignored:
+                ignored += sql_ignored
+
+        # And analysse the reverse SQL
+        if runsql.reversible and runsql.reverse_sql != RunSQL.noop:
+            sql_statements = []
+            if isinstance(runsql.reverse_sql, (list, tuple)):
+                for sql in runsql.reverse_sql:
+                    params = None
+                    if isinstance(sql, (list, tuple)):
+                        elements = len(sql)
+                        if elements == 2:
+                            sql, params = sql
+                        else:
+                            raise ValueError("Expected a 2-tuple but got %d" % elements)
+                        sql_statements.append(sql % params)
+                    else:
+                        sql_statements.append(sql)
+            else:
+                sql_statements.append(runsql.reverse_sql)
+
+            sql_errors, sql_ignored = analyse_sql_statements(
+                sql_statements,
+                settings.DATABASES[self.database]["ENGINE"],
+                self.exclude_migration_tests,
+            )
+            if sql_errors:
+                error += sql_errors
+            if sql_ignored:
+                ignored += sql_ignored
+
+        return error, ignored, warning
diff --git a/docs/incompatibilities.md b/docs/incompatibilities.md
index 5df319b..0818d3b 100644
--- a/docs/incompatibilities.md
+++ b/docs/incompatibilities.md
@@ -19,3 +19,23 @@ We are happy to add more if you can specify them to us.
 On data migrations, the linter will show a warning when:
 * you are missing a reverse migration
 * the RunPython arguments do not respect the naming convention `apps, schema_editor`
+* the model variable name is different from the model class name in the `get_model` call
+
+## Codes
+
+You can ignore check through the `--exclude-migration-test` option and specifying any of the codes:
+
+|               Code                  |            Description                                                                          |
+|-------------------------------------|----------------------------------------------------------------------|
+|`NOT_NULL`                           | Not NULL constraint on columns
+|`DROP_COLUMN`                        | Dropping columns
+|`DROP_TABLE`                         | Dropping tables
+|`RENAME_COLUMN`                      | Renaming columns
+|`RENAME_TABLE`                       | Renaming tables
+|`ALTER_COLUMN`                       | Altering columns (could be backward compatible)
+|`ADD_UNIQUE`                         | Add unique constraints
+|`REVERSIBLE_DATA_MIGRATION`          | RunPython data migration is not reversible (missing reverse code)
+|`NAMING_CONVENTION_RUNPYTHON_ARGS`   | By convention, RunPython names two arguments: apps, schema_editor
+|`DATA_MIGRATION_MODEL_IMPORT`        | Missing apps.get_model() calls for model
+|`DATA_MIGRATION_MODEL_VARIABLE_NAME` | The model variable name is different from the model class itself
+|`REVERSIBLE_RUNSQL_DATA_MIGRATION`   | RunSQL data migration is not reversible (missing reverse SQL)


Test patch:
diff --git a/tests/functional/test_data_migrations.py b/tests/functional/test_data_migrations.py
index 2405091..d048b3d 100644
--- a/tests/functional/test_data_migrations.py
+++ b/tests/functional/test_data_migrations.py
@@ -2,6 +2,7 @@ import os
 import unittest
 
 from django.conf import settings
+from django.db import migrations
 
 from django_migration_linter import MigrationLinter
 from tests import fixtures
@@ -66,13 +67,15 @@ class DataMigrationDetectionTestCase(unittest.TestCase):
         self.assertEqual(1, self.linter.nb_erroneous)
         self.assertTrue(self.linter.has_errors)
 
+
+class DataMigrationModelImportTestCase(unittest.TestCase):
     def test_missing_get_model_import(self):
         def incorrect_importing_model_forward(apps, schema_editor):
             from tests.test_project.app_data_migrations.models import MyModel
 
             MyModel.objects.filter(id=1).first()
 
-        issues = MigrationLinter.get_data_migration_model_import_issues(
+        issues = MigrationLinter.get_runpython_model_import_issues(
             incorrect_importing_model_forward
         )
         self.assertEqual(1, len(issues))
@@ -83,11 +86,193 @@ class DataMigrationDetectionTestCase(unittest.TestCase):
             MyVeryLongLongLongModel = apps.get_model(
                 "app_data_migrations", "MyVeryLongLongLongModel"
             )
+            MultiLineModel = apps.get_model(
+                "app_data_migrations",
+                "MultiLineModel",
+            )
 
             MyModel.objects.filter(id=1).first()
             MyVeryLongLongLongModel.objects.filter(id=1).first()
+            MultiLineModel.objects.all()
 
-        issues = MigrationLinter.get_data_migration_model_import_issues(
+        issues = MigrationLinter.get_runpython_model_import_issues(
             correct_importing_model_forward
         )
         self.assertEqual(0, len(issues))
+
+    def test_not_overlapping_model_name(self):
+        """
+        Correct for the import error, but should raise a warning
+        """
+
+        def forward_method(apps, schema_editor):
+            User = apps.get_model("auth", "CustomUserModel")
+
+            User.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_import_issues(forward_method)
+        self.assertEqual(0, len(issues))
+
+    def test_correct_one_param_get_model_import(self):
+        def forward_method(apps, schema_editor):
+            User = apps.get_model("auth.User")
+
+            User.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_import_issues(forward_method)
+        self.assertEqual(0, len(issues))
+
+    def test_not_overlapping_one_param(self):
+        """
+        Not an error, but should raise a warning
+        """
+
+        def forward_method(apps, schema_editor):
+            User = apps.get_model("auth.CustomUserModel")
+
+            User.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_import_issues(forward_method)
+        self.assertEqual(0, len(issues))
+
+
+class DataMigrationModelVariableNamingTestCase(unittest.TestCase):
+    def test_same_variable_name(self):
+        def forward_op(apps, schema_editor):
+            MyModel = apps.get_model("app", "MyModel")
+
+            MyModel.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_variable_naming_issues(forward_op)
+        self.assertEqual(0, len(issues))
+
+    def test_same_variable_name_multiline(self):
+        def forward_op(apps, schema_editor):
+            MyModelVeryLongLongLongLongLong = apps.get_model(
+                "app", "MyModelVeryLongLongLongLongLong"
+            )
+
+            MyModelVeryLongLongLongLongLong.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_variable_naming_issues(forward_op)
+        self.assertEqual(0, len(issues))
+
+    def test_same_variable_name_multiline2(self):
+        def forward_op(apps, schema_editor):
+            MyModelVeryLongLongLongLongLong = apps.get_model(
+                "app_name_longlonglonglongapp",
+                "MyModelVeryLongLongLongLongLong",
+            )
+
+            MyModelVeryLongLongLongLongLong.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_variable_naming_issues(forward_op)
+        self.assertEqual(0, len(issues))
+
+    def test_different_variable_name(self):
+        def forward_op(apps, schema_editor):
+            some_model = apps.get_model("app", "MyModel")
+
+            some_model.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_variable_naming_issues(forward_op)
+        self.assertEqual(1, len(issues))
+
+    def test_diff_variable_name_multiline(self):
+        def forward_op(apps, schema_editor):
+            MyModelVeryLongLongLongLongLongNot = apps.get_model(
+                "app", "MyModelVeryLongLongLongLongLong"
+            )
+
+            MyModelVeryLongLongLongLongLongNot.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_variable_naming_issues(forward_op)
+        self.assertEqual(1, len(issues))
+
+    def test_diff_variable_name_multiline2(self):
+        def forward_op(apps, schema_editor):
+            MyModelVeryLongLongLongLongLongNot = apps.get_model(
+                "app_name_longlonglonglongapp",
+                "MyModelVeryLongLongLongLongLong",
+            )
+
+            MyModelVeryLongLongLongLongLongNot.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_variable_naming_issues(forward_op)
+        self.assertEqual(1, len(issues))
+
+    def test_same_variable_name_one_param(self):
+        def forward_op(apps, schema_editor):
+            MyModel = apps.get_model("app.MyModel")
+
+            MyModel.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_variable_naming_issues(forward_op)
+        self.assertEqual(0, len(issues))
+
+    def test_different_variable_name_one_param(self):
+        def forward_op(apps, schema_editor):
+            mymodel = apps.get_model("app.MyModel")
+
+            mymodel.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_variable_naming_issues(forward_op)
+        self.assertEqual(1, len(issues))
+
+    def test_correct_variable_name_one_param_multiline(self):
+        def forward_op(apps, schema_editor):
+            AVeryLongModelName = apps.get_model(
+                "quite_long_app_name.AVeryLongModelName"
+            )
+
+            AVeryLongModelName.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_variable_naming_issues(forward_op)
+        self.assertEqual(0, len(issues))
+
+    def test_different_variable_name_one_param_multiline(self):
+        def forward_op(apps, schema_editor):
+            m = apps.get_model(
+                "quite_long_app_name_name_name.AVeryLongModelNameNameName"
+            )
+
+            m.objects.filter(id=1).first()
+
+        issues = MigrationLinter.get_runpython_model_variable_naming_issues(forward_op)
+        self.assertEqual(1, len(issues))
+
+
+class RunSQLMigrationTestCase(unittest.TestCase):
+    def setUp(self):
+        test_project_path = os.path.dirname(settings.BASE_DIR)
+        self.linter = MigrationLinter(
+            test_project_path,
+            include_apps=fixtures.DATA_MIGRATIONS,
+        )
+
+    def test_missing_reserve_migration(self):
+        runsql = migrations.RunSQL("sql;")
+
+        error, ignored, warning = self.linter.lint_runsql(runsql)
+        self.assertEqual("REVERSIBLE_RUNSQL_DATA_MIGRATION", warning[0]["code"])
+
+    def test_sql_linting_error(self):
+        runsql = migrations.RunSQL("ALTER TABLE t DROP COLUMN t;")
+
+        error, ignored, warning = self.linter.lint_runsql(runsql)
+        self.assertEqual("DROP_COLUMN", error[0]["code"])
+
+    def test_sql_linting_error_array(self):
+        runsql = migrations.RunSQL(
+            ["ALTER TABLE t DROP COLUMN c;", "ALTER TABLE t RENAME COLUMN c;"]
+        )
+
+        error, ignored, warning = self.linter.lint_runsql(runsql)
+        self.assertEqual("DROP_COLUMN", error[0]["code"])
+        self.assertEqual("RENAME_COLUMN", error[1]["code"])
+
+    def test_sql_linting_error_args(self):
+        runsql = migrations.RunSQL([("ALTER TABLE %s DROP COLUMN %s;", ("t", "c"))])
+
+        error, ignored, warning = self.linter.lint_runsql(runsql)
+        self.assertEqual("DROP_COLUMN", error[0]["code"])