innov
/
django1
1
0
Fork 0
Code Issues Pull Requests 1 Projects Releases Wiki Activity

Added test coverage for DatabaseOperations.sql_flush().

This commit is contained in:
Jon Dufresne 2020-04-15 02:20:46 -07:00 committed by Mariusz Felisiak
parent eeab63e57e
commit 8bcca47e83
5 changed files with 416 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
tests/backends/base/test_operations.py
View File

@ -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)

62
tests/backends/mysql/test_operations.py Normal file
View File

@ -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;')

133
tests/backends/oracle/test_operations.py
View File

@ -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])

84
tests/backends/postgresql/test_operations.py Normal file
View File

@ -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"',
],
)

94
tests/backends/sqlite/test_operations.py Normal file
View File

@ -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";',
],
)