Refs #25530 -- Tracked references of deferred SQL statements.

This commit is contained in:
Simon Charette 2016-05-24 15:25:05 -04:00
parent ea91ad4c13
commit 3b429c9673
6 changed files with 302 additions and 32 deletions

View file

@ -0,0 +1,125 @@
from django.db.backends.ddl_references import (
Columns, ForeignKeyName, IndexName, Statement, Table,
)
from django.test import SimpleTestCase
class TableTests(SimpleTestCase):
def setUp(self):
self.reference = Table('table', lambda table: table.upper())
def test_references_table(self):
self.assertIs(self.reference.references_table('table'), True)
self.assertIs(self.reference.references_table('other'), False)
def test_repr(self):
self.assertEqual(repr(self.reference), "<Table 'TABLE'>")
def test_str(self):
self.assertEqual(str(self.reference), 'TABLE')
class ColumnsTests(TableTests):
def setUp(self):
self.reference = Columns(
'table', ['first_column', 'second_column'], lambda column: column.upper()
)
def test_references_column(self):
self.assertIs(self.reference.references_column('other', 'first_column'), False)
self.assertIs(self.reference.references_column('table', 'third_column'), False)
self.assertIs(self.reference.references_column('table', 'first_column'), True)
def test_repr(self):
self.assertEqual(repr(self.reference), "<Columns 'FIRST_COLUMN, SECOND_COLUMN'>")
def test_str(self):
self.assertEqual(str(self.reference), 'FIRST_COLUMN, SECOND_COLUMN')
class IndexNameTests(ColumnsTests):
def setUp(self):
def create_index_name(table_name, column_names, suffix):
return ', '.join("%s_%s_%s" % (table_name, column_name, suffix) for column_name in column_names)
self.reference = IndexName(
'table', ['first_column', 'second_column'], 'suffix', create_index_name
)
def test_repr(self):
self.assertEqual(repr(self.reference), "<IndexName 'table_first_column_suffix, table_second_column_suffix'>")
def test_str(self):
self.assertEqual(str(self.reference), 'table_first_column_suffix, table_second_column_suffix')
class ForeignKeyNameTests(IndexNameTests):
def setUp(self):
def create_foreign_key_name(table_name, column_names, suffix):
return ', '.join("%s_%s_%s" % (table_name, column_name, suffix) for column_name in column_names)
self.reference = ForeignKeyName(
'table', ['first_column', 'second_column'],
'to_table', ['to_first_column', 'to_second_column'],
'%(to_table)s_%(to_column)s_fk',
create_foreign_key_name,
)
def test_references_table(self):
super().test_references_table()
self.assertIs(self.reference.references_table('to_table'), True)
def test_references_column(self):
super().test_references_column()
self.assertIs(self.reference.references_column('to_table', 'second_column'), False)
self.assertIs(self.reference.references_column('to_table', 'to_second_column'), True)
def test_repr(self):
self.assertEqual(
repr(self.reference),
"<ForeignKeyName 'table_first_column_to_table_to_first_column_fk, "
"table_second_column_to_table_to_first_column_fk'>"
)
def test_str(self):
self.assertEqual(
str(self.reference),
'table_first_column_to_table_to_first_column_fk, '
'table_second_column_to_table_to_first_column_fk'
)
class MockReference(object):
def __init__(self, representation, referenced_tables, referenced_columns):
self.representation = representation
self.referenced_tables = referenced_tables
self.referenced_columns = referenced_columns
def references_table(self, table):
return table in self.referenced_tables
def references_column(self, table, column):
return (table, column) in self.referenced_columns
def __str__(self):
return self.representation
class StatementTests(SimpleTestCase):
def test_references_table(self):
statement = Statement('', reference=MockReference('', {'table'}, {}), non_reference='')
self.assertIs(statement.references_table('table'), True)
self.assertIs(statement.references_table('other'), False)
def test_references_column(self):
statement = Statement('', reference=MockReference('', {}, {('table', 'column')}), non_reference='')
self.assertIs(statement.references_column('table', 'column'), True)
self.assertIs(statement.references_column('other', 'column'), False)
def test_repr(self):
reference = MockReference('reference', {}, {})
statement = Statement("%(reference)s - %(non_reference)s", reference=reference, non_reference='non_reference')
self.assertEqual(repr(statement), "<Statement 'reference - non_reference'>")
def test_str(self):
reference = MockReference('reference', {}, {})
statement = Statement("%(reference)s - %(non_reference)s", reference=reference, non_reference='non_reference')
self.assertEqual(str(statement), 'reference - non_reference')

View file

@ -51,7 +51,7 @@ class SchemaIndexesTests(TestCase):
def test_index_together(self):
editor = connection.schema_editor()
index_sql = editor._model_indexes_sql(Article)
index_sql = [str(statement) for statement in editor._model_indexes_sql(Article)]
self.assertEqual(len(index_sql), 1)
# Ensure the index name is properly quoted
self.assertIn(
@ -70,7 +70,7 @@ class SchemaIndexesTests(TestCase):
def test_postgresql_text_indexes(self):
"""Test creation of PostgreSQL-specific text indexes (#12234)"""
from .models import IndexedArticle
index_sql = connection.schema_editor()._model_indexes_sql(IndexedArticle)
index_sql = [str(statement) for statement in connection.schema_editor()._model_indexes_sql(IndexedArticle)]
self.assertEqual(len(index_sql), 5)
self.assertIn('("headline" varchar_pattern_ops)', index_sql[1])
self.assertIn('("body" text_pattern_ops)', index_sql[3])
@ -99,7 +99,7 @@ class SchemaIndexesMySQLTests(TransactionTestCase):
)
if storage != "InnoDB":
self.skip("This test only applies to the InnoDB storage engine")
index_sql = connection.schema_editor()._model_indexes_sql(ArticleTranslation)
index_sql = [str(statement) for statement in connection.schema_editor()._model_indexes_sql(ArticleTranslation)]
self.assertEqual(index_sql, [
'CREATE INDEX `indexes_articletranslation_article_no_constraint_id_d6c0806b` '
'ON `indexes_articletranslation` (`article_no_constraint_id`)'
@ -114,7 +114,7 @@ class SchemaIndexesMySQLTests(TransactionTestCase):
new_field.set_attributes_from_name('new_foreign_key')
editor.add_field(ArticleTranslation, new_field)
field_created = True
self.assertEqual(editor.deferred_sql, [
self.assertEqual([str(statement) for statement in editor.deferred_sql], [
'ALTER TABLE `indexes_articletranslation` '
'ADD CONSTRAINT `indexes_articletrans_new_foreign_key_id_d27a9146_fk_indexes_a` '
'FOREIGN KEY (`new_foreign_key_id`) REFERENCES `indexes_article` (`id`)'

View file

@ -15,7 +15,7 @@ def sql_for_table(model):
def sql_for_index(model):
return '\n'.join(connection.schema_editor()._model_indexes_sql(model))
return '\n'.join(str(sql) for sql in connection.schema_editor()._model_indexes_sql(model))
# We can't test the DEFAULT_TABLESPACE and DEFAULT_INDEX_TABLESPACE settings