from __future__ import annotations import tempfile import unittest from unittest.mock import patch from django.db import ProgrammingError, migrations, models from django.db.migrations import Migration from django.db.migrations.operations.base import Operation from django_migration_linter import MigrationLinter from django_migration_linter.sql_analyser import ( analyse_sql_statements, get_sql_analyser_class, ) class LinterFunctionsTestCase(unittest.TestCase): def test_get_sql(self): linter = MigrationLinter() sql_statements = linter.get_sql("app_add_not_null_column", "0001") self.assertEqual(len(sql_statements), 6) self.assertEqual(sql_statements[0], "BEGIN;") self.assertEqual(sql_statements[-1], "COMMIT;") def test_has_errors(self): linter = MigrationLinter(database="mysql") self.assertFalse(linter.has_errors) m = Migration("0001_create_table", "app_add_not_null_column") linter.lint_migration(m) self.assertFalse(linter.has_errors) m = Migration("0002_add_new_not_null_field", "app_add_not_null_column") linter.lint_migration(m) self.assertTrue(linter.has_errors) m = Migration("0001_create_table", "app_add_not_null_column") linter.lint_migration(m) self.assertTrue(linter.has_errors) def test_ignore_migration_include_apps(self): linter = MigrationLinter(include_apps=("app_add_not_null_column",)) self.assertTrue(linter.should_ignore_migration("app_correct", "0001")) self.assertTrue(linter.should_ignore_migration("app_correct", "0002")) self.assertFalse( linter.should_ignore_migration("app_add_not_null_column", "0001") ) def test_ignore_migration_exclude_apps(self): linter = MigrationLinter(exclude_apps=("app_add_not_null_column",)) self.assertFalse(linter.should_ignore_migration("app_correct", "0001")) self.assertFalse(linter.should_ignore_migration("app_correct", "0002")) self.assertTrue( linter.should_ignore_migration("app_add_not_null_column", "0001") ) def test_ignore_migration_name_contains(self): linter = MigrationLinter(ignore_name_contains="foo") self.assertFalse(linter.should_ignore_migration("app_correct", "0001_initial")) self.assertTrue(linter.should_ignore_migration("app_correct", "0002_foo")) def test_ignore_migration_full_name(self): linter = MigrationLinter(ignore_name=("0002_foo",)) self.assertFalse(linter.should_ignore_migration("app_correct", "0001_initial")) self.assertTrue(linter.should_ignore_migration("app_correct", "0002_foo")) def test_include_migration_name_contains(self): linter = MigrationLinter(include_name_contains="foo") self.assertTrue(linter.should_ignore_migration("app_correct", "0001_initial")) self.assertFalse(linter.should_ignore_migration("app_correct", "0002_foo")) def test_include_migration_full_name(self): linter = MigrationLinter(include_name=("0002_foo",)) self.assertTrue(linter.should_ignore_migration("app_correct", "0001_initial")) self.assertFalse(linter.should_ignore_migration("app_correct", "0002_foo")) def test_include_migration_multiple_names(self): linter = MigrationLinter(include_name=("0002_foo", "0003_bar")) self.assertTrue(linter.should_ignore_migration("app_correct", "0001_initial")) self.assertFalse(linter.should_ignore_migration("app_correct", "0002_foo")) self.assertFalse(linter.should_ignore_migration("app_correct", "0003_bar")) def test_gather_all_migrations(self): linter = MigrationLinter() migrations = linter._gather_all_migrations() self.assertGreater(len(list(migrations)), 1) def test_ignore_unapplied_migrations(self): linter = MigrationLinter(only_applied_migrations=True) linter.migration_loader.applied_migrations = {("app_correct", "0002_foo")} self.assertTrue(linter.should_ignore_migration("app_correct", "0001_initial")) self.assertFalse(linter.should_ignore_migration("app_correct", "0002_foo")) def test_ignore_applied_migrations(self): linter = MigrationLinter(only_unapplied_migrations=True) linter.migration_loader.applied_migrations = {("app_correct", "0002_foo")} self.assertFalse(linter.should_ignore_migration("app_correct", "0001_initial")) self.assertTrue(linter.should_ignore_migration("app_correct", "0002_foo")) def test_exclude_migration_tests(self): m = Migration("0002_add_new_not_null_field", "app_add_not_null_column") linter = MigrationLinter(exclude_migration_tests=[], database="mysql") linter.lint_migration(m) self.assertTrue(linter.has_errors) linter = MigrationLinter(exclude_migration_tests=["NOT_NULL"], database="mysql") linter.lint_migration(m) self.assertFalse(linter.has_errors) def test_read_migrations_unknown_file(self): file_path = "unknown_file" with self.assertRaises(Exception): MigrationLinter.read_migrations_list(file_path) @patch( "django_migration_linter.migration_linter.call_command", side_effect=ProgrammingError, ) def test_raise_exception_on_sqlmigrate_error(self, call_command_mock): linter = MigrationLinter( exclude_migration_tests=[], database="mysql", ignore_sqlmigrate_errors=False ) with self.assertRaises(ProgrammingError): linter.get_sql("app_correct", "0002_foo") @patch( "django_migration_linter.migration_linter.call_command", side_effect=ProgrammingError, ) def test_ignore_exception_on_sqlmigrate_error(self, call_command_mock): linter = MigrationLinter( exclude_migration_tests=[], database="mysql", ignore_sqlmigrate_errors=True ) sql_result = linter.get_sql("app_correct", "0002_foo") self.assertEqual([], sql_result) def test_read_migrations_no_file(self): migration_list = MigrationLinter.read_migrations_list(None) self.assertIsNone(migration_list) def test_read_migrations_empty_file(self): with tempfile.NamedTemporaryFile() as tmp: migration_list = MigrationLinter.read_migrations_list(tmp.name) self.assertEqual([], migration_list) def test_read_migrations_from_file(self): tmp = tempfile.NamedTemporaryFile(mode="w", delete=False) tmp.write( "test_project/app_add_not_null_column/migrations/0001_create_table.py\n" ) tmp.write("unknown\n") tmp.write( "test_project/app_add_not_null_column/migrations/0002_add_new_not_null_field.py\n" ) tmp.close() migration_list = MigrationLinter.read_migrations_list(tmp.name) self.assertEqual( [ ("app_add_not_null_column", "0001_create_table"), ("app_add_not_null_column", "0002_add_new_not_null_field"), ], migration_list, ) def test_gather_migrations_with_list(self): linter = MigrationLinter() migrations = linter._gather_all_migrations( migrations_list=[ ("app_add_not_null_column", "0001_create_table"), ("app_add_not_null_column", "0002_add_new_not_null_field"), ] ) self.assertEqual(2, len(list(migrations))) def test_ignore_initial_migrations(self): linter = MigrationLinter(ignore_initial_migrations=True) self.assertTrue( linter.should_ignore_migration( "app_correct", "0001_initial", is_initial=True ) ) self.assertFalse( linter.should_ignore_migration("app_correct", "0002_foo", is_initial=False) ) def test_initial_migration_still_lints_runsql_and_runpython(self): class CteInsertOperation(Operation): reduces_to_sql = True reversible = True def state_forwards(self, app_label, state): return None def database_forwards(self, app_label, schema_editor, from_state, to_state): schema_editor.execute( 'WITH inserted AS (SELECT 1) INSERT INTO "archive_capture" ("id") SELECT * FROM inserted;' ) def database_backwards( self, app_label, schema_editor, from_state, to_state ): return None def describe(self): return "Insert an archive capture through a CTE" def forward(apps, schema_editor): return None def unsafe_forward(apps, schema_editor): Legacy = None Legacy.objects.all() migration = Migration("0001_initial", "app_initial_operations") migration.initial = True migration.operations = [ migrations.RunSQL("DROP TABLE legacy_data;"), migrations.RunPython(forward), ] linter = MigrationLinter.__new__(MigrationLinter) linter.exclude_migration_tests = [] linter.sql_analyser_class = get_sql_analyser_class("postgresql") linter.include_apps = None linter.exclude_apps = None linter.ignore_name_contains = None linter.include_name_contains = None linter.ignore_name = [] linter.include_name = [] linter.only_applied_migrations = False linter.only_unapplied_migrations = False linter.ignore_initial_migrations = False errors, _, warnings = linter.analyse_data_migration(migration) self.assertFalse( linter.should_ignore_migration( migration.app_label, migration.name, is_initial=migration.initial ) ) self.assertTrue(linter.has_unknown_data_operations(migration.operations)) declarative_migration = Migration("0001_initial", "app_declarative_schema") declarative_migration.initial = True declarative_migration.operations = [ migrations.CreateModel(name="ArchiveCapture", fields=[]), migrations.AddField( model_name="archivecapture", name="source_id", field=models.IntegerField(), ), migrations.AddConstraint( model_name="archivecapture", constraint=models.UniqueConstraint( fields=("source_id",), name="archive_source" ), ), ] self.assertFalse( linter.has_unknown_data_operations(declarative_migration.operations) ) renamed_migration = Migration("0001_initial", "app_renamed_schema") renamed_migration.initial = True renamed_migration.operations = [ migrations.CreateModel(name="ArchiveCapture", fields=[]), migrations.RenameModel( old_name="archivecapture", new_name="archivedcapture" ), migrations.AddField( model_name="archivecapture", name="source_id", field=models.IntegerField(), ), ] self.assertTrue( linter.has_unknown_data_operations(renamed_migration.operations) ) self.assertEqual(["DROP_TABLE"], [issue.code for issue in errors]) self.assertEqual( ["RUNSQL_REVERSIBLE", "RUNPYTHON_REVERSIBLE"], [issue.code for issue in warnings], ) nested_migration = Migration("0001_nested_initial", "app_nested_operations") nested_migration.initial = True nested_migration.operations = [ migrations.SeparateDatabaseAndState( database_operations=[ migrations.RunSQL("DROP TABLE legacy_data;"), migrations.RunPython(unsafe_forward), ] ) ] errors, _, warnings = linter.analyse_data_migration(nested_migration) self.assertEqual( ["DROP_TABLE", "RUNPYTHON_MODEL_IMPORT"], [issue.code for issue in errors], ) self.assertEqual( [ "RUNSQL_REVERSIBLE", "RUNPYTHON_REVERSIBLE", "RUNPYTHON_MODEL_VARIABLE_NAME", ], [issue.code for issue in warnings], ) migration = Migration("0001_initial", "app_initial_schema_and_data") migration.initial = True migration.operations = [ migrations.CreateModel(name="ArchiveCapture", fields=[]), migrations.SeparateDatabaseAndState( database_operations=[migrations.RunPython(forward)] ), migrations.AddField( model_name="archivecapture", name="source_id", field=models.IntegerField(), ), ] self.assertTrue(linter.has_unknown_data_operations(migration.operations)) linter.django_path = None linter.no_cache = True linter.all_warnings_as_errors = False linter.warnings_as_errors_tests = None linter.no_output = True linter.quiet = [] linter.reset_counters() linter.get_migration_hash = lambda *args: "initial-schema-and-data" linter.get_sql = lambda *args: [ 'CREATE TABLE "archive_capture" ("id" uuid NOT NULL PRIMARY KEY);', 'ALTER TABLE "archive_capture" ADD COLUMN "source_id" uuid NOT NULL REFERENCES "archive_source" ("id");', ] linter.analyse_data_migration = lambda *args: ([], [], []) linter.lint_migration(migration) self.assertTrue(linter.has_errors) migration.operations[1] = migrations.SeparateDatabaseAndState( database_operations=[CteInsertOperation()] ) linter.get_sql = lambda *args: [ 'CREATE TABLE "archive_capture" ("id" uuid NOT NULL PRIMARY KEY);', 'WITH inserted AS (SELECT 1) INSERT INTO "archive_capture" ("id") SELECT * FROM inserted;', 'ALTER TABLE "archive_capture" ADD COLUMN "source_id" uuid NOT NULL REFERENCES "archive_source" ("id");', ] linter.reset_counters() self.assertTrue(linter.has_unknown_data_operations(migration.operations)) errors, _, _ = analyse_sql_statements( linter.sql_analyser_class, linter.get_sql(), has_unknown_data_operations=True, ) self.assertEqual(["NOT_NULL"], [issue.code for issue in errors]) linter.lint_migration(migration) self.assertTrue(linter.has_errors) migration.operations[1] = migrations.SeparateDatabaseAndState( database_operations=[ migrations.RunSQL("INSERT INTO archive_capture VALUES (1);") ] ) linter.reset_counters() linter.lint_migration(migration) self.assertTrue(linter.has_errors)