Fixed #24529 -- Allowed double squashing of migrations.

Co-authored-by: Raphael Gaschignard <raphael@rtpg.co>

tests/migrations/test_commands.py

@@ -2,6 +2,7 @@ import datetime
 import importlib
 import io
 import os
+import re
 import shutil
 import sys
 from pathlib import Path
@@ -28,7 +29,9 @@ from django.db.backends.base.schema import BaseDatabaseSchemaEditor
 from django.db.backends.utils import truncate_name
 from django.db.migrations.autodetector import MigrationAutodetector
 from django.db.migrations.exceptions import InconsistentMigrationHistory
+from django.db.migrations.loader import MigrationLoader
 from django.db.migrations.recorder import MigrationRecorder
+from django.db.migrations.writer import MigrationWriter
 from django.test import TestCase, override_settings, skipUnlessDBFeature
 from django.test.utils import captured_stdout, extend_sys_path, isolate_apps
 from django.utils import timezone
@@ -2939,6 +2942,137 @@ class SquashMigrationsTests(MigrationTestBase):
             "  you can delete them.\n" % squashed_migration_file,
         )
 
+    def test_squashmigrations_replacement_cycle(self):
+        out = io.StringIO()
+        with self.temporary_migration_module(
+            module="migrations.test_migrations_squashed_loop"
+        ):
+            # Hits a squash replacement cycle check error, but the actual failure is
+            # dependent on the order in which the files are read on disk.
+            with self.assertRaisesRegex(
+                CommandError,
+                r"Cyclical squash replacement found, starting at"
+                r" \('migrations', '2_(squashed|auto)'\)",
+            ):
+                call_command(
+                    "migrate", "migrations", "--plan", interactive=False, stdout=out
+                )
+
+    def test_squashmigrations_squashes_already_squashed(self):
+        out = io.StringIO()
+
+        with self.temporary_migration_module(
+            module="migrations.test_migrations_squashed_complex"
+        ):
+            call_command(
+                "squashmigrations",
+                "migrations",
+                "3_squashed_5",
+                "--squashed-name",
+                "double_squash",
+                stdout=out,
+                interactive=False,
+            )
+
+            loader = MigrationLoader(connection)
+            migration = loader.disk_migrations[("migrations", "0001_double_squash")]
+            # Confirm the replaces mechanism holds the squashed migration
+            # (and not what it squashes, as the squash operations are what
+            # end up being used).
+            self.assertEqual(
+                migration.replaces,
+                [
+                    ("migrations", "1_auto"),
+                    ("migrations", "2_auto"),
+                    ("migrations", "3_squashed_5"),
+                ],
+            )
+
+            out = io.StringIO()
+            call_command(
+                "migrate", "migrations", "--plan", interactive=False, stdout=out
+            )
+
+            migration_plan = re.findall("migrations.(.+)\n", out.getvalue())
+            self.assertEqual(migration_plan, ["0001_double_squash", "6_auto", "7_auto"])
+
+    def test_squash_partially_applied(self):
+        """
+        Replacement migrations are partially applied. Then we squash again and
+        verify that only unapplied migrations will be applied by "migrate".
+        """
+        out = io.StringIO()
+
+        with self.temporary_migration_module(
+            module="migrations.test_migrations_squashed_partially_applied"
+        ):
+            # Apply first 2 migrations.
+            call_command("migrate", "migrations", "0002", interactive=False, stdout=out)
+
+            # Squash the 2 migrations, that we just applied + 1 more.
+            call_command(
+                "squashmigrations",
+                "migrations",
+                "0001",
+                "0003",
+                "--squashed-name",
+                "squashed_0001_0003",
+                stdout=out,
+                interactive=False,
+            )
+
+            # Update the 4th migration to depend on the squash(replacement) migration.
+            loader = MigrationLoader(connection)
+            migration = loader.disk_migrations[
+                ("migrations", "0004_remove_mymodel1_field_1_mymodel1_field_3_and_more")
+            ]
+            migration.dependencies = [("migrations", "0001_squashed_0001_0003")]
+            writer = MigrationWriter(migration)
+            with open(writer.path, "w", encoding="utf-8") as fh:
+                fh.write(writer.as_string())
+
+            # Squash the squash(replacement) migration with the 4th migration.
+            call_command(
+                "squashmigrations",
+                "migrations",
+                "0001_squashed_0001_0003",
+                "0004",
+                "--squashed-name",
+                "squashed_0001_0004",
+                stdout=out,
+                interactive=False,
+            )
+
+            loader = MigrationLoader(connection)
+            migration = loader.disk_migrations[
+                ("migrations", "0001_squashed_0001_0004")
+            ]
+            self.assertEqual(
+                migration.replaces,
+                [
+                    ("migrations", "0001_squashed_0001_0003"),
+                    (
+                        "migrations",
+                        "0004_remove_mymodel1_field_1_mymodel1_field_3_and_more",
+                    ),
+                ],
+            )
+
+            # Verify that only unapplied migrations will be applied.
+            out = io.StringIO()
+            call_command(
+                "migrate", "migrations", "--plan", interactive=False, stdout=out
+            )
+
+            migration_plan = re.findall("migrations.(.+)\n", out.getvalue())
+            self.assertEqual(
+                migration_plan,
+                [
+                    "0003_alter_mymodel2_unique_together",
+                    "0004_remove_mymodel1_field_1_mymodel1_field_3_and_more",
+                ],
+            )
+
     def test_squashmigrations_initial_attribute(self):
         with self.temporary_migration_module(
             module="migrations.test_migrations"