Fixed #30054 -- Implemented cascaded flush on SQLite.

This is required to maintain foreign key integrity when using
TransactionTestCase.available_apps.

Refs #30033, #14204, #20483.

1 files changed, 21 insertions, 6 deletions

diff --git a/django/db/backends/sqlite3/operations.py b/django/db/backends/sqlite3/operations.py
index cd71afb23b..858e10ef84 100644
--- a/django/db/backends/sqlite3/operations.py
+++ b/django/db/backends/sqlite3/operations.py
@@ -159,6 +159,27 @@ class DatabaseOperations(BaseDatabaseOperations):
         return -1
 
     def sql_flush(self, style, tables, sequences, allow_cascade=False):
+        if tables and allow_cascade:
+            # Simulate TRUNCATE CASCADE by recursively collecting the tables
+            # referencing the tables to be flushed.
+            query = """
+            WITH tables AS (
+                %s
+                UNION
+                SELECT sqlite_master.name
+                FROM sqlite_master
+                JOIN tables ON (
+                    sql REGEXP %%s || tables.name || %%s
+                )
+            ) SELECT name FROM tables;
+            """ % ' UNION '.join("SELECT '%s' name" % table for table in tables)
+            params = (
+                r'(?i)\s+references\s+("|\')?',
+                r'("|\')?\s*\(',
+            )
+            with self.connection.cursor() as cursor:
+                results = cursor.execute(query, params)
+                tables = [row[0] for row in results.fetchall()]
         sql = ['%s %s %s;' % (
             style.SQL_KEYWORD('DELETE'),
             style.SQL_KEYWORD('FROM'),
@@ -168,12 +189,6 @@ class DatabaseOperations(BaseDatabaseOperations):
         # sql_flush() implementations). Just return SQL at this point
         return sql
 
-    def execute_sql_flush(self, using, sql_list):
-        # To prevent possible violation of foreign key constraints, deactivate
-        # constraints outside of the transaction created in super().
-        with self.connection.constraint_checks_disabled():
-            super().execute_sql_flush(using, sql_list)
-
     def adapt_datetimefield_value(self, value):
         if value is None:
             return None