Work on SQL Analyser module

* With database specific functions
* Functions that analyse the entire transaction
* Unit tests that only test this module

django_migration_linter/migration_linter.py

@@ -20,6 +20,7 @@ import os
 import re
 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
 
@@ -122,7 +123,9 @@ class MigrationLinter(object):
         sql_statements = self.get_sql(app_label, migration_name)
         exclude_migration_tests = self.exclude_migration_tests or []
         errors, ignored = analyse_sql_statements(
-            sql_statements, exclude_migration_tests
+            sql_statements,
+            settings.DATABASES[self.database]["ENGINE"],
+            exclude_migration_tests,
         )
 
         if not errors:

django_migration_linter/sql_analyser.py

@@ -14,63 +14,141 @@
 
 import re
 import logging
+from copy import deepcopy
 
 logger = logging.getLogger(__name__)
 
 
-def has_default(sql, **kwargs):
-    if re.search("SET DEFAULT", sql) and kwargs["errors"]:
-        err = next((err for err in kwargs["errors"] if err["code"] == "NOT_NULL"), None)
-        if err:
-            logger.info(
-                (
-                    "Found a NOT_NULL error in migration, "
-                    "but it has a default value added: {}"
-                ).format(err)
-            )
-            kwargs["errors"].remove(err)
-
-    return False  # Never fails
+def find_error_dict_with_code(tests, code):
+    return next((test_dict for test_dict in tests if test_dict["code"] == code), None)
 
 
-migration_tests = (
+def has_not_null_column(sql_statements, **kwargs):
+    # TODO: improve to detect that the same column is concerned
+    return any(
+        re.search("(?<!DROP )NOT NULL", sql) and not sql.startswith("CREATE TABLE")
+        for sql in sql_statements
+    ) and not any("SET DEFAULT" in sql for sql in sql_statements)
+
+
+base_migration_tests = [
+    {
+        "code": "RENAME_TABLE",
+        "fn": lambda sql, **kw: re.search("RENAME TABLE", sql)
+        or re.search("ALTER TABLE .* RENAME TO", sql),
+        "err_msg": "RENAMING tables",
+        "mode": "one_liner",
+    },
     {
         "code": "NOT_NULL",
-        "fn": lambda sql, **kw: re.search("(?<!DROP )NOT NULL", sql)
-        and not re.search("CREATE TABLE", sql),
+        "fn": has_not_null_column,
         "err_msg": "NOT NULL constraint on columns",
+        "mode": "transaction",
     },
     {
         "code": "DROP_COLUMN",
         "fn": lambda sql, **kw: re.search("DROP COLUMN", sql),
         "err_msg": "DROPPING columns",
+        "mode": "one_liner",
     },
     {
         "code": "RENAME_COLUMN",
         "fn": lambda sql, **kw: re.search("ALTER TABLE .* CHANGE", sql)
         or re.search("ALTER TABLE .* RENAME COLUMN", sql),
         "err_msg": "RENAMING columns",
-    },
-    {
-        "code": "RENAME_TABLE",
-        "fn": lambda sql, **kw: re.search("RENAME TABLE", sql)
-        or re.search("ALTER TABLE .* RENAME TO", sql),
-        "err_msg": "RENAMING tables",
+        "mode": "one_liner",
     },
     {
         "code": "ALTER_COLUMN",
-        "fn": lambda sql, **kw: re.search("ALTER TABLE .* MODIFY", sql)
-        or re.search("ALTER TABLE .* ALTER COLUMN .* TYPE", sql),
+        "fn": lambda sql, **kw: re.search("ALTER TABLE .* ALTER COLUMN .* TYPE", sql),
         "err_msg": (
             "ALTERING columns (Could be backward compatible. "
             "You may ignore this migration.)"
         ),
+        "mode": "one_liner",
     },
-    {"code": "", "fn": has_default, "err_msg": ""},
-)
+]
+
+mysql_migration_tests = [
+    {
+        "code": "ALTER_COLUMN",
+        "fn": lambda sql, **kw: re.search("ALTER TABLE .* MODIFY .* (?!NULL);?$", sql),
+        "mode": "one_liner",
+    }
+]
+
+postgresql_migration_tests = []
+
+sqlite_migration_tests = [
+    {
+        "code": "RENAME_TABLE",
+        "fn": lambda sql, **kw: re.search("ALTER TABLE .* RENAME TO", sql)
+        and "__old" not in sql,
+    },
+    {
+        "code": "NOT_NULL",
+        "fn": lambda sql_statements, **kw: any(
+            re.search("NOT NULL(?! PRIMARY)(?! DEFAULT)", sql) for sql in sql_statements
+        )
+        and any(
+            re.search("ALTER TABLE .* RENAME TO", sql) and "__old" in sql
+            for sql in sql_statements
+        ),
+        "mode": "transaction",
+    },
+]
 
 
-def build_error_dict(migration_test, sql_statement):
+def update_migration_tests(base_tests, specific_tests):
+    for override_test in specific_tests:
+        migration_test_dict = find_error_dict_with_code(
+            base_tests, override_test["code"]
+        )
+
+        if migration_test_dict is None or not override_test["code"]:
+            migration_test_dict = {}
+            base_tests.append(migration_test_dict)
+
+        for key in override_test.keys():
+            migration_test_dict[key] = override_test[key]
+    return base_tests
+
+
+def get_migration_tests(database_vendor):
+    migration_tests = deepcopy(base_migration_tests)
+
+    if "mysql" in database_vendor:
+        migration_tests = update_migration_tests(migration_tests, mysql_migration_tests)
+    elif "postgre" in database_vendor:
+        migration_tests = update_migration_tests(
+            migration_tests, postgresql_migration_tests
+        )
+    elif "sqlite" in database_vendor:
+        migration_tests = update_migration_tests(
+            migration_tests, sqlite_migration_tests
+        )
+
+    return migration_tests
+
+
+def get_single_line_migration_tests(database_vendor):
+    return [
+        test
+        for test in get_migration_tests(database_vendor)
+        if test["mode"] == "one_liner"
+    ]
+
+
+def get_transaction_migration_tests(database_vendor):
+    # TODO: improve transaction to detect BEGIN & COMMIT
+    return [
+        test
+        for test in get_migration_tests(database_vendor)
+        if test["mode"] == "transaction"
+    ]
+
+
+def build_error_dict(migration_test, sql_statement=""):
     table_search = re.search("TABLE `([^`]*)`", sql_statement, re.IGNORECASE)
     col_search = re.search("COLUMN `([^`]*)`", sql_statement, re.IGNORECASE)
     return {
@@ -81,10 +159,13 @@ def build_error_dict(migration_test, sql_statement):
     }
 
 
-def analyse_sql_statements(sql_statements, exclude_migration_tests):
+def analyse_sql_statements(
+    sql_statements, database_vendor, exclude_migration_tests=None
+):
+    exclude_migration_tests = exclude_migration_tests or []
     errors, ignored = [], []
     for statement in sql_statements:
-        for test in migration_tests:
+        for test in get_single_line_migration_tests(database_vendor):
             if test["fn"](statement, errors=errors):
                 if test["code"] in exclude_migration_tests:
                     logger.debug("Testing {0} -- IGNORED".format(statement))
@@ -96,4 +177,18 @@ def analyse_sql_statements(sql_statements, exclude_migration_tests):
                     errors.append(err)
             else:
                 logger.debug("Testing {0} -- PASSED".format(statement))
+
+    for test in get_transaction_migration_tests(database_vendor):
+        if test["fn"](sql_statements, errors=errors):
+            if test["code"] in exclude_migration_tests:
+                logger.debug("Testing {0} -- IGNORED".format(sql_statements))
+                err = build_error_dict(migration_test=test)
+                ignored.append(err)
+            else:
+                logger.debug("Testing {0} -- ERROR".format(sql_statements))
+                err = build_error_dict(migration_test=test)
+                errors.append(err)
+        else:
+            logger.debug("Testing {0} -- PASSED".format(sql_statements))
+
     return errors, ignored