diff --git a/tests/backends/base/test_operations.py b/tests/backends/base/test_operations.py index 1ba7ae099e2..5dff48d44e1 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 00000000000..f1d9342a7b2 --- /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 bcae17cec63..97760ecbfed 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 00000000000..b073f688f45 --- /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 00000000000..34c4d823dad --- /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";', + ], + )