Added test coverage for DatabaseOperations.sql_flush().

5 files changed, 416 insertions, 2 deletions

diff --git a/tests/backends/base/test_operations.py b/tests/backends/base/test_operations.py
index 1ba7ae099e..5dff48d44e 100644
--- a/tests/backends/base/test_operations.py
+++ b/tests/backends/base/test_operations.py
@@ -1,13 +1,17 @@
 import decimal
 
-from django.db import NotSupportedError, connection
+from django.core.management.color import no_style
+from django.db import NotSupportedError, connection, transaction
 from django.db.backends.base.operations import BaseDatabaseOperations
 from django.db.models import DurationField
 from django.test import (
-    SimpleTestCase, TestCase, override_settings, skipIfDBFeature,
+    SimpleTestCase, TestCase, TransactionTestCase, override_settings,
+    skipIfDBFeature,
 )
 from django.utils import timezone
 
+from ..models import Author, Book
+
 
 class SimpleDatabaseOperationTests(SimpleTestCase):
     may_require_msg = 'subclasses of BaseDatabaseOperations may require a %s() method'
@@ -144,3 +148,40 @@ class DatabaseOperationTests(TestCase):
         )
         with self.assertRaisesMessage(NotSupportedError, msg):
             self.ops.subtract_temporals(duration_field_internal_type, None, None)
+
+
+class SqlFlushTests(TransactionTestCase):
+    available_apps = ['backends']
+
+    def test_sql_flush_no_tables(self):
+        self.assertEqual(connection.ops.sql_flush(no_style(), [], []), [])
+
+    def test_execute_sql_flush_statements(self):
+        with transaction.atomic():
+            author = Author.objects.create(name='George Orwell')
+            Book.objects.create(author=author)
+            author = Author.objects.create(name='Harper Lee')
+            Book.objects.create(author=author)
+            Book.objects.create(author=author)
+            self.assertIs(Author.objects.exists(), True)
+            self.assertIs(Book.objects.exists(), True)
+
+        sql_list = connection.ops.sql_flush(
+            no_style(),
+            [Author._meta.db_table, Book._meta.db_table],
+            [
+                {
+                    'table': Author._meta.db_table,
+                    'column': Author._meta.pk.db_column,
+                },
+            ],
+            allow_cascade=True,
+        )
+        connection.ops.execute_sql_flush(connection.alias, sql_list)
+
+        with transaction.atomic():
+            self.assertIs(Author.objects.exists(), False)
+            self.assertIs(Book.objects.exists(), False)
+            if connection.features.supports_sequence_reset:
+                author = Author.objects.create(name='F. Scott Fitzgerald')
+                self.assertEqual(author.pk, 1)
diff --git a/tests/backends/mysql/test_operations.py b/tests/backends/mysql/test_operations.py
new file mode 100644
index 0000000000..f1d9342a7b
--- /dev/null
+++ b/tests/backends/mysql/test_operations.py
@@ -0,0 +1,62 @@
+import unittest
+
+from django.core.management.color import no_style
+from django.db import connection
+from django.test import SimpleTestCase
+
+from ..models import Person, Square, Tag
+
+
+@unittest.skipUnless(connection.vendor == 'mysql', 'MySQL tests.')
+class MySQLOperationsTests(SimpleTestCase):
+    def test_sql_flush(self):
+        # allow_cascade doesn't change statements on MySQL.
+        for allow_cascade in [False, True]:
+            with self.subTest(allow_cascade=allow_cascade):
+                statements = connection.ops.sql_flush(
+                    no_style(),
+                    [Person._meta.db_table, Tag._meta.db_table],
+                    [],
+                    allow_cascade=allow_cascade,
+                )
+                self.assertEqual(statements[0], 'SET FOREIGN_KEY_CHECKS = 0;')
+                # The tables are processed in an unordered set.
+                self.assertEqual(
+                    sorted(statements[1:-1]),
+                    [
+                        'DELETE FROM `backends_person`;',
+                        'DELETE FROM `backends_tag`;',
+                    ],
+                )
+                self.assertEqual(statements[-1], 'SET FOREIGN_KEY_CHECKS = 1;')
+
+    def test_sql_flush_sequences(self):
+        # allow_cascade doesn't change statements on MySQL.
+        for allow_cascade in [False, True]:
+            with self.subTest(allow_cascade=allow_cascade):
+                statements = connection.ops.sql_flush(
+                    no_style(),
+                    [Person._meta.db_table, Square._meta.db_table, Tag._meta.db_table],
+                    [
+                        {
+                            'table': Person._meta.db_table,
+                            'column': Person._meta.pk.db_column,
+                        },
+                        {
+                            'table': Tag._meta.db_table,
+                            'column': Tag._meta.pk.db_column,
+                        },
+                    ],
+                    allow_cascade=allow_cascade,
+                )
+                self.assertEqual(statements[0], 'SET FOREIGN_KEY_CHECKS = 0;')
+                # The tables are processed in an unordered set.
+                self.assertEqual(
+                    sorted(statements[1:-1]),
+                    [
+                        'DELETE FROM `backends_square`;',
+                        'TRUNCATE `backends_person`;',
+                        'TRUNCATE `backends_tag`;',
+                    ],
+                )
+                self.assertEqual(statements[-1], 'SET FOREIGN_KEY_CHECKS = 1;')
diff --git a/tests/backends/oracle/test_operations.py b/tests/backends/oracle/test_operations.py
index bcae17cec6..97760ecbfe 100644
--- a/tests/backends/oracle/test_operations.py
+++ b/tests/backends/oracle/test_operations.py
@@ -1,7 +1,10 @@
 import unittest
 
+from django.core.management.color import no_style
 from django.db import connection
 
+from ..models import Person, Tag
+
 
 @unittest.skipUnless(connection.vendor == 'oracle', 'Oracle tests')
 class OperationsTests(unittest.TestCase):
@@ -23,3 +26,133 @@ class OperationsTests(unittest.TestCase):
             connection.ops.bulk_batch_size(['id', 'other'], objects),
             connection.features.max_query_params // 2,
         )
+
+    def test_sql_flush(self):
+        statements = connection.ops.sql_flush(
+            no_style(),
+            [Person._meta.db_table, Tag._meta.db_table],
+            [],
+        )
+        # The tables and constraints are processed in an unordered set.
+        self.assertEqual(
+            statements[0],
+            'ALTER TABLE "BACKENDS_TAG" DISABLE CONSTRAINT '
+            '"BACKENDS__CONTENT_T_FD9D7A85_F" KEEP INDEX;',
+        )
+        self.assertEqual(
+            sorted(statements[1:-1]),
+            [
+                'TRUNCATE TABLE "BACKENDS_PERSON";',
+                'TRUNCATE TABLE "BACKENDS_TAG";',
+            ],
+        )
+        self.assertEqual(
+            statements[-1],
+            'ALTER TABLE "BACKENDS_TAG" ENABLE CONSTRAINT '
+            '"BACKENDS__CONTENT_T_FD9D7A85_F";',
+        )
+
+    def test_sql_flush_allow_cascade(self):
+        statements = connection.ops.sql_flush(
+            no_style(),
+            [Person._meta.db_table, Tag._meta.db_table],
+            [],
+            allow_cascade=True,
+        )
+        # The tables and constraints are processed in an unordered set.
+        self.assertEqual(
+            statements[0],
+            'ALTER TABLE "BACKENDS_VERYLONGMODELNAME540F" DISABLE CONSTRAINT '
+            '"BACKENDS__PERSON_ID_1DD5E829_F" KEEP INDEX;',
+        )
+        self.assertEqual(
+            sorted(statements[1:-1]),
+            [
+                'TRUNCATE TABLE "BACKENDS_PERSON";',
+                'TRUNCATE TABLE "BACKENDS_TAG";',
+                'TRUNCATE TABLE "BACKENDS_VERYLONGMODELNAME540F";',
+            ],
+        )
+        self.assertEqual(
+            statements[-1],
+            'ALTER TABLE "BACKENDS_VERYLONGMODELNAME540F" ENABLE CONSTRAINT '
+            '"BACKENDS__PERSON_ID_1DD5E829_F";',
+        )
+
+    def test_sql_flush_sequences(self):
+        statements = connection.ops.sql_flush(
+            no_style(),
+            [Person._meta.db_table, Tag._meta.db_table],
+            [
+                {
+                    'table': Person._meta.db_table,
+                    'column': Person._meta.pk.db_column,
+                },
+                {
+                    'table': Tag._meta.db_table,
+                    'column': Tag._meta.pk.db_column,
+                },
+            ],
+        )
+        # The tables and constraints are processed in an unordered set.
+        self.assertEqual(
+            statements[0],
+            'ALTER TABLE "BACKENDS_TAG" DISABLE CONSTRAINT '
+            '"BACKENDS__CONTENT_T_FD9D7A85_F" KEEP INDEX;',
+        )
+        self.assertEqual(
+            sorted(statements[1:3]),
+            [
+                'TRUNCATE TABLE "BACKENDS_PERSON";',
+                'TRUNCATE TABLE "BACKENDS_TAG";',
+            ],
+        )
+        self.assertEqual(
+            statements[3],
+            'ALTER TABLE "BACKENDS_TAG" ENABLE CONSTRAINT '
+            '"BACKENDS__CONTENT_T_FD9D7A85_F";',
+        )
+        # Sequences.
+        self.assertEqual(len(statements[4:]), 2)
+        self.assertIn('BACKENDS_PERSON_SQ', statements[4])
+        self.assertIn('BACKENDS_TAG_SQ', statements[5])
+
+    def test_sql_flush_sequences_allow_cascade(self):
+        statements = connection.ops.sql_flush(
+            no_style(),
+            [Person._meta.db_table, Tag._meta.db_table],
+            [
+                {
+                    'table': Person._meta.db_table,
+                    'column': Person._meta.pk.db_column,
+                },
+                {
+                    'table': Tag._meta.db_table,
+                    'column': Tag._meta.pk.db_column,
+                },
+            ],
+            allow_cascade=True,
+        )
+        # The tables and constraints are processed in an unordered set.
+        self.assertEqual(
+            statements[0],
+            'ALTER TABLE "BACKENDS_VERYLONGMODELNAME540F" DISABLE CONSTRAINT '
+            '"BACKENDS__PERSON_ID_1DD5E829_F" KEEP INDEX;',
+        )
+        self.assertEqual(
+            sorted(statements[1:4]),
+            [
+                'TRUNCATE TABLE "BACKENDS_PERSON";',
+                'TRUNCATE TABLE "BACKENDS_TAG";',
+                'TRUNCATE TABLE "BACKENDS_VERYLONGMODELNAME540F";',
+            ],
+        )
+        self.assertEqual(
+            statements[4],
+            'ALTER TABLE "BACKENDS_VERYLONGMODELNAME540F" ENABLE CONSTRAINT '
+            '"BACKENDS__PERSON_ID_1DD5E829_F";',
+        )
+        # Sequences.
+        self.assertEqual(len(statements[5:]), 2)
+        self.assertIn('BACKENDS_PERSON_SQ', statements[5])
+        self.assertIn('BACKENDS_TAG_SQ', statements[6])
diff --git a/tests/backends/postgresql/test_operations.py b/tests/backends/postgresql/test_operations.py
new file mode 100644
index 0000000000..b073f688f4
--- /dev/null
+++ b/tests/backends/postgresql/test_operations.py
@@ -0,0 +1,84 @@
+import unittest
+
+from django.core.management.color import no_style
+from django.db import connection
+from django.test import SimpleTestCase
+
+from ..models import Person, Tag
+
+
+@unittest.skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests.')
+class PostgreSQLOperationsTests(SimpleTestCase):
+    def test_sql_flush(self):
+        self.assertEqual(
+            connection.ops.sql_flush(
+                no_style(),
+                [Person._meta.db_table, Tag._meta.db_table],
+                [],
+            ),
+            ['TRUNCATE "backends_person", "backends_tag";'],
+        )
+
+    def test_sql_flush_allow_cascade(self):
+        self.assertEqual(
+            connection.ops.sql_flush(
+                no_style(),
+                [Person._meta.db_table, Tag._meta.db_table],
+                [],
+                allow_cascade=True,
+            ),
+            ['TRUNCATE "backends_person", "backends_tag" CASCADE;'],
+        )
+
+    def test_sql_flush_sequences(self):
+        sequence_reset_sql = (
+            "SELECT setval(pg_get_serial_sequence('%s','id'), 1, false);"
+        )
+        self.assertEqual(
+            connection.ops.sql_flush(
+                no_style(),
+                [Person._meta.db_table, Tag._meta.db_table],
+                [
+                    {
+                        'table': Person._meta.db_table,
+                        'column': Person._meta.pk.db_column,
+                    },
+                    {
+                        'table': Tag._meta.db_table,
+                        'column': Tag._meta.pk.db_column,
+                    },
+                ],
+            ),
+            [
+                'TRUNCATE "backends_person", "backends_tag";',
+                sequence_reset_sql % '"backends_person"',
+                sequence_reset_sql % '"backends_tag"',
+            ],
+        )
+
+    def test_sql_flush_sequences_allow_cascade(self):
+        sequence_reset_sql = (
+            "SELECT setval(pg_get_serial_sequence('%s','id'), 1, false);"
+        )
+        self.assertEqual(
+            connection.ops.sql_flush(
+                no_style(),
+                [Person._meta.db_table, Tag._meta.db_table],
+                [
+                    {
+                        'table': Person._meta.db_table,
+                        'column': Person._meta.pk.db_column,
+                    },
+                    {
+                        'table': Tag._meta.db_table,
+                        'column': Tag._meta.pk.db_column,
+                    },
+                ],
+                allow_cascade=True,
+            ),
+            [
+                'TRUNCATE "backends_person", "backends_tag" CASCADE;',
+                sequence_reset_sql % '"backends_person"',
+                sequence_reset_sql % '"backends_tag"',
+            ],
+        )
diff --git a/tests/backends/sqlite/test_operations.py b/tests/backends/sqlite/test_operations.py
new file mode 100644
index 0000000000..34c4d823da
--- /dev/null
+++ b/tests/backends/sqlite/test_operations.py
@@ -0,0 +1,94 @@
+import unittest
+
+from django.core.management.color import no_style
+from django.db import connection
+from django.test import TestCase
+
+from ..models import Person, Tag
+
+
+@unittest.skipUnless(connection.vendor == 'sqlite', 'SQLite tests.')
+class SQLiteOperationsTests(TestCase):
+    def test_sql_flush(self):
+        self.assertEqual(
+            connection.ops.sql_flush(
+                no_style(),
+                [Person._meta.db_table, Tag._meta.db_table],
+                [],
+            ),
+            [
+                'DELETE FROM "backends_person";',
+                'DELETE FROM "backends_tag";',
+            ],
+        )
+
+    def test_sql_flush_allow_cascade(self):
+        statements = connection.ops.sql_flush(
+            no_style(),
+            [Person._meta.db_table, Tag._meta.db_table],
+            [],
+            allow_cascade=True,
+        )
+        self.assertEqual(
+            # The tables are processed in an unordered set.
+            sorted(statements),
+            [
+                'DELETE FROM "backends_person";',
+                'DELETE FROM "backends_tag";',
+                'DELETE FROM "backends_verylongmodelnamezzzzzzzzzzzzzzzzzzzzzz'
+                'zzzzzzzzzzzzzzzzzzzz_m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzz'
+                'zzzzzzzzzzzzzzzzzzzzzzz";',
+            ],
+        )
+
+    def test_sql_flush_sequences(self):
+        # sequences doesn't change statements on SQLite.
+        self.assertEqual(
+            connection.ops.sql_flush(
+                no_style(),
+                [Person._meta.db_table, Tag._meta.db_table],
+                [
+                    {
+                        'table': Person._meta.db_table,
+                        'column': Person._meta.pk.db_column,
+                    },
+                    {
+                        'table': Tag._meta.db_table,
+                        'column': Tag._meta.pk.db_column,
+                    },
+                ],
+            ),
+            [
+                'DELETE FROM "backends_person";',
+                'DELETE FROM "backends_tag";',
+            ],
+        )
+
+    def test_sql_flush_sequences_allow_cascade(self):
+        # sequences doesn't change statements on SQLite.
+        statements = connection.ops.sql_flush(
+            no_style(),
+            [Person._meta.db_table, Tag._meta.db_table],
+            [
+                {
+                    'table': Person._meta.db_table,
+                    'column': Person._meta.pk.db_column,
+                },
+                {
+                    'table': Tag._meta.db_table,
+                    'column': Tag._meta.pk.db_column,
+                },
+            ],
+            allow_cascade=True,
+        )
+        self.assertEqual(
+            # The tables are processed in an unordered set.
+            sorted(statements),
+            [
+                'DELETE FROM "backends_person";',
+                'DELETE FROM "backends_tag";',
+                'DELETE FROM "backends_verylongmodelnamezzzzzzzzzzzzzzzzzzzzzz'
+                'zzzzzzzzzzzzzzzzzzzz_m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzz'
+                'zzzzzzzzzzzzzzzzzzzzzzz";',
+            ],
+        )