public/django-migration-linter
15
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: invalidate empty-table proof after writes and table renames

Browse Source
This commit is contained in:
ftrinidad committed 2026-10-28 18:43:04 -03:00
1 parent c7a4376d35
commit 66f39b1c48
7 files changed
+285 -29
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/unit/test_linter.py
+159 -16
View File
@@ -4,12 +4,15 @@ import tempfile
import unittest
from unittest.mock import patch
from django.db import ProgrammingError
from django.db import migrations
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 get_sql_analyser_class
from django_migration_linter.sql_analyser import (
analyse_sql_statements,
get_sql_analyser_class,
)
class LinterFunctionsTestCase(unittest.TestCase):
@@ -79,18 +82,7 @@ class LinterFunctionsTestCase(unittest.TestCase):
self.assertFalse(linter.should_ignore_migration("app_correct", "0003_bar"))
def test_gather_all_migrations(self):
linter = MigrationLinter.__new__(MigrationLinter)
linter.exclude_migration_tests = []
linter.sql_analyser_class = get_sql_analyser_class("sqlite")
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
linter = MigrationLinter()
migrations = linter._gather_all_migrations()
self.assertGreater(len(list(migrations)), 1)
@@ -197,9 +189,33 @@ class LinterFunctionsTestCase(unittest.TestCase):
)
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 = [
@@ -208,7 +224,7 @@ class LinterFunctionsTestCase(unittest.TestCase):
]
linter = MigrationLinter.__new__(MigrationLinter)
linter.exclude_migration_tests = []
linter.sql_analyser_class = get_sql_analyser_class("sqlite")
linter.sql_analyser_class = get_sql_analyser_class("postgresql")
linter.include_apps = None
linter.exclude_apps = None
linter.ignore_name_contains = None
@@ -226,8 +242,135 @@ class LinterFunctionsTestCase(unittest.TestCase):
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)
tests/unit/test_sql_analyser.py
+27 -1
View File
@@ -11,12 +11,13 @@ from django_migration_linter.sql_analyser import (
class SqlAnalyserTestCase(unittest.TestCase):
database_vendor = "default"
def analyse_sql(self, sql):
def analyse_sql(self, sql, **kwargs):
if isinstance(sql, str):
sql = sql.splitlines()
return analyse_sql_statements(
get_sql_analyser_class(self.database_vendor),
sql_statements=sql,
**kwargs,
)
def assertValidSql(self, sql, allow_warnings=False):
@@ -249,6 +250,31 @@ class PostgresqlAnalyserTestCase(SqlAnalyserTestCase):
]
self.assertValidSql(sql)
def test_add_required_foreign_key_after_insert_into_new_table(self):
sql = [
'CREATE TABLE "archive_capture" ("id" uuid NOT NULL PRIMARY KEY);',
'INSERT INTO "archive_capture" ("id") VALUES (gen_random_uuid());',
'ALTER TABLE "archive_capture" ADD COLUMN "source_id" uuid NOT NULL REFERENCES "archive_source" ("id") DEFERRABLE INITIALLY DEFERRED;',
]
self.assertBackwardIncompatibleSql(sql, "NOT_NULL")
def test_add_required_foreign_key_after_unknown_data_operation(self):
sql = [
'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") DEFERRABLE INITIALLY DEFERRED;',
]
errors, _, _ = self.analyse_sql(sql, has_unknown_data_operations=True)
self.assertTrue(any(error.code == "NOT_NULL" for error in errors))
def test_add_required_foreign_key_after_table_name_reuse(self):
sql = [
'CREATE TABLE "foo" ("id" uuid NOT NULL PRIMARY KEY);',
'ALTER TABLE "foo" RENAME TO "baz";',
'ALTER TABLE "bar" RENAME TO "foo";',
'ALTER TABLE "foo" ADD COLUMN "source_id" uuid NOT NULL REFERENCES "archive_source" ("id");',
]
self.assertBackwardIncompatibleSql(sql, "NOT_NULL")
def test_alter_column(self):
sql = 'ALTER TABLE "app_alter_column_a" ALTER COLUMN "field" TYPE varchar(10) USING "field"::varchar(10);'
self.assertBackwardIncompatibleSql(sql)