import datetime from unittest import skipIf, skipUnless from django.db import connection from django.db.models import CASCADE, ForeignKey, Index, Q from django.test import ( TestCase, TransactionTestCase, skipIfDBFeature, skipUnlessDBFeature, ) from django.test.utils import override_settings from django.utils import timezone from .models import ( Article, ArticleTranslation, IndexedArticle2, IndexTogetherSingleList, ) class SchemaIndexesTests(TestCase): """ Test index handling by the db.backends.schema infrastructure. """ def test_index_name_hash(self): """ Index names should be deterministic. """ editor = connection.schema_editor() index_name = editor._create_index_name( table_name=Article._meta.db_table, column_names=("c1",), suffix="123", ) self.assertEqual(index_name, "indexes_article_c1_a52bd80b123") def test_index_name(self): """ Index names on the built-in database backends:: * Are truncated as needed. * Include all the column names. * Include a deterministic hash. """ long_name = 'l%sng' % ('o' * 100) editor = connection.schema_editor() index_name = editor._create_index_name( table_name=Article._meta.db_table, column_names=('c1', 'c2', long_name), suffix='ix', ) expected = { 'mysql': 'indexes_article_c1_c2_looooooooooooooooooo_255179b2ix', 'oracle': 'indexes_a_c1_c2_loo_255179b2ix', 'postgresql': 'indexes_article_c1_c2_loooooooooooooooooo_255179b2ix', 'sqlite': 'indexes_article_c1_c2_l%sng_255179b2ix' % ('o' * 100), } if connection.vendor not in expected: self.skipTest('This test is only supported on the built-in database backends.') self.assertEqual(index_name, expected[connection.vendor]) def test_index_together(self): editor = connection.schema_editor() 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( connection.ops.quote_name( editor._create_index_name(Article._meta.db_table, ['headline', 'pub_date'], suffix='_idx') ), index_sql[0] ) def test_index_together_single_list(self): # Test for using index_together with a single list (#22172) index_sql = connection.schema_editor()._model_indexes_sql(IndexTogetherSingleList) self.assertEqual(len(index_sql), 1) def test_columns_list_sql(self): index = Index(fields=['headline'], name='whitespace_idx') editor = connection.schema_editor() self.assertIn( '(%s)' % editor.quote_name('headline'), str(index.create_sql(Article, editor)), ) def test_descending_columns_list_sql(self): index = Index(fields=['-headline'], name='whitespace_idx') editor = connection.schema_editor() self.assertIn( '(%s DESC)' % editor.quote_name('headline'), str(index.create_sql(Article, editor)), ) @skipIf(connection.vendor == 'postgresql', 'opclasses are PostgreSQL only') class SchemaIndexesNotPostgreSQLTests(TransactionTestCase): available_apps = ['indexes'] def test_create_index_ignores_opclasses(self): index = Index( name='test_ops_class', fields=['headline'], opclasses=['varchar_pattern_ops'], ) with connection.schema_editor() as editor: # This would error if opclasses weren't ignored. editor.add_index(IndexedArticle2, index) # The `condition` parameter is ignored by databases that don't support partial # indexes. @skipIfDBFeature('supports_partial_indexes') class PartialIndexConditionIgnoredTests(TransactionTestCase): available_apps = ['indexes'] def test_condition_ignored(self): index = Index( name='test_condition_ignored', fields=['published'], condition=Q(published=True), ) with connection.schema_editor() as editor: # This would error if condition weren't ignored. editor.add_index(Article, index) self.assertNotIn( 'WHERE %s' % editor.quote_name('published'), str(index.create_sql(Article, editor)) ) @skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests') class SchemaIndexesPostgreSQLTests(TransactionTestCase): available_apps = ['indexes'] get_opclass_query = ''' SELECT opcname, c.relname FROM pg_opclass AS oc JOIN pg_index as i on oc.oid = ANY(i.indclass) JOIN pg_class as c on c.oid = i.indexrelid WHERE c.relname = '%s' ''' def test_text_indexes(self): """Test creation of PostgreSQL-specific text indexes (#12234)""" from .models import 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]) # unique=True and db_index=True should only create the varchar-specific # index (#19441). self.assertIn('("slug" varchar_pattern_ops)', index_sql[4]) def test_virtual_relation_indexes(self): """Test indexes are not created for related objects""" index_sql = connection.schema_editor()._model_indexes_sql(Article) self.assertEqual(len(index_sql), 1) def test_ops_class(self): index = Index( name='test_ops_class', fields=['headline'], opclasses=['varchar_pattern_ops'], ) with connection.schema_editor() as editor: editor.add_index(IndexedArticle2, index) with editor.connection.cursor() as cursor: cursor.execute(self.get_opclass_query % 'test_ops_class') self.assertEqual(cursor.fetchall(), [('varchar_pattern_ops', 'test_ops_class')]) def test_ops_class_multiple_columns(self): index = Index( name='test_ops_class_multiple', fields=['headline', 'body'], opclasses=['varchar_pattern_ops', 'text_pattern_ops'], ) with connection.schema_editor() as editor: editor.add_index(IndexedArticle2, index) with editor.connection.cursor() as cursor: cursor.execute(self.get_opclass_query % 'test_ops_class_multiple') expected_ops_classes = ( ('varchar_pattern_ops', 'test_ops_class_multiple'), ('text_pattern_ops', 'test_ops_class_multiple'), ) self.assertCountEqual(cursor.fetchall(), expected_ops_classes) def test_ops_class_partial(self): index = Index( name='test_ops_class_partial', fields=['body'], opclasses=['text_pattern_ops'], condition=Q(headline__contains='China'), ) with connection.schema_editor() as editor: editor.add_index(IndexedArticle2, index) with editor.connection.cursor() as cursor: cursor.execute(self.get_opclass_query % 'test_ops_class_partial') self.assertCountEqual(cursor.fetchall(), [('text_pattern_ops', 'test_ops_class_partial')]) def test_ops_class_partial_tablespace(self): indexname = 'test_ops_class_tblspace' index = Index( name=indexname, fields=['body'], opclasses=['text_pattern_ops'], condition=Q(headline__contains='China'), db_tablespace='pg_default', ) with connection.schema_editor() as editor: editor.add_index(IndexedArticle2, index) self.assertIn('TABLESPACE "pg_default" ', str(index.create_sql(IndexedArticle2, editor))) with editor.connection.cursor() as cursor: cursor.execute(self.get_opclass_query % indexname) self.assertCountEqual(cursor.fetchall(), [('text_pattern_ops', indexname)]) def test_ops_class_descending(self): indexname = 'test_ops_class_ordered' index = Index( name=indexname, fields=['-body'], opclasses=['text_pattern_ops'], ) with connection.schema_editor() as editor: editor.add_index(IndexedArticle2, index) with editor.connection.cursor() as cursor: cursor.execute(self.get_opclass_query % indexname) self.assertCountEqual(cursor.fetchall(), [('text_pattern_ops', indexname)]) def test_ops_class_descending_partial(self): indexname = 'test_ops_class_ordered_partial' index = Index( name=indexname, fields=['-body'], opclasses=['text_pattern_ops'], condition=Q(headline__contains='China'), ) with connection.schema_editor() as editor: editor.add_index(IndexedArticle2, index) with editor.connection.cursor() as cursor: cursor.execute(self.get_opclass_query % indexname) self.assertCountEqual(cursor.fetchall(), [('text_pattern_ops', indexname)]) def test_ops_class_columns_lists_sql(self): index = Index( fields=['headline'], name='whitespace_idx', opclasses=['text_pattern_ops'], ) with connection.schema_editor() as editor: self.assertIn( '(%s text_pattern_ops)' % editor.quote_name('headline'), str(index.create_sql(Article, editor)), ) def test_ops_class_descending_columns_list_sql(self): index = Index( fields=['-headline'], name='whitespace_idx', opclasses=['text_pattern_ops'], ) with connection.schema_editor() as editor: self.assertIn( '(%s text_pattern_ops DESC)' % editor.quote_name('headline'), str(index.create_sql(Article, editor)), ) @skipUnless(connection.vendor == 'mysql', 'MySQL tests') class SchemaIndexesMySQLTests(TransactionTestCase): available_apps = ['indexes'] def test_no_index_for_foreignkey(self): """ MySQL on InnoDB already creates indexes automatically for foreign keys. (#14180). An index should be created if db_constraint=False (#26171). """ with connection.cursor() as cursor: storage = connection.introspection.get_storage_engine( cursor, ArticleTranslation._meta.db_table, ) if storage != "InnoDB": self.skip("This test only applies to the InnoDB storage engine") 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`)' ]) # The index also shouldn't be created if the ForeignKey is added after # the model was created. field_created = False try: with connection.schema_editor() as editor: new_field = ForeignKey(Article, CASCADE) new_field.set_attributes_from_name('new_foreign_key') editor.add_field(ArticleTranslation, new_field) field_created = True # No deferred SQL. The FK constraint is included in the # statement to add the field. self.assertFalse(editor.deferred_sql) finally: if field_created: with connection.schema_editor() as editor: editor.remove_field(ArticleTranslation, new_field) @skipUnlessDBFeature('supports_partial_indexes') # SQLite doesn't support timezone-aware datetimes when USE_TZ is False. @override_settings(USE_TZ=True) class PartialIndexTests(TransactionTestCase): # Schema editor is used to create the index to test that it works. available_apps = ['indexes'] def test_partial_index(self): with connection.schema_editor() as editor: index = Index( name='recent_article_idx', fields=['pub_date'], condition=Q( pub_date__gt=datetime.datetime( year=2015, month=1, day=1, # PostgreSQL would otherwise complain about the lookup # being converted to a mutable function (by removing # the timezone in the cast) which is forbidden. tzinfo=timezone.get_current_timezone(), ), ) ) self.assertIn( 'WHERE %s' % editor.quote_name('pub_date'), str(index.create_sql(Article, schema_editor=editor)) ) editor.add_index(index=index, model=Article) with connection.cursor() as cursor: self.assertIn(index.name, connection.introspection.get_constraints( cursor=cursor, table_name=Article._meta.db_table, )) editor.remove_index(index=index, model=Article) def test_integer_restriction_partial(self): with connection.schema_editor() as editor: index = Index( name='recent_article_idx', fields=['id'], condition=Q(pk__gt=1), ) self.assertIn( 'WHERE %s' % editor.quote_name('id'), str(index.create_sql(Article, schema_editor=editor)) ) editor.add_index(index=index, model=Article) with connection.cursor() as cursor: self.assertIn(index.name, connection.introspection.get_constraints( cursor=cursor, table_name=Article._meta.db_table, )) editor.remove_index(index=index, model=Article) def test_boolean_restriction_partial(self): with connection.schema_editor() as editor: index = Index( name='published_index', fields=['published'], condition=Q(published=True), ) self.assertIn( 'WHERE %s' % editor.quote_name('published'), str(index.create_sql(Article, schema_editor=editor)) ) editor.add_index(index=index, model=Article) with connection.cursor() as cursor: self.assertIn(index.name, connection.introspection.get_constraints( cursor=cursor, table_name=Article._meta.db_table, )) editor.remove_index(index=index, model=Article) @skipUnlessDBFeature('supports_functions_in_partial_indexes') def test_multiple_conditions(self): with connection.schema_editor() as editor: index = Index( name='recent_article_idx', fields=['pub_date', 'headline'], condition=( Q(pub_date__gt=datetime.datetime( year=2015, month=1, day=1, tzinfo=timezone.get_current_timezone(), )) & Q(headline__contains='China') ), ) sql = str(index.create_sql(Article, schema_editor=editor)) where = sql.find('WHERE') self.assertIn( 'WHERE (%s' % editor.quote_name('pub_date'), sql ) # Because each backend has different syntax for the operators, # check ONLY the occurrence of headline in the SQL. self.assertGreater(sql.rfind('headline'), where) editor.add_index(index=index, model=Article) with connection.cursor() as cursor: self.assertIn(index.name, connection.introspection.get_constraints( cursor=cursor, table_name=Article._meta.db_table, )) editor.remove_index(index=index, model=Article) def test_is_null_condition(self): with connection.schema_editor() as editor: index = Index( name='recent_article_idx', fields=['pub_date'], condition=Q(pub_date__isnull=False), ) self.assertIn( 'WHERE %s IS NOT NULL' % editor.quote_name('pub_date'), str(index.create_sql(Article, schema_editor=editor)) ) editor.add_index(index=index, model=Article) with connection.cursor() as cursor: self.assertIn(index.name, connection.introspection.get_constraints( cursor=cursor, table_name=Article._meta.db_table, )) editor.remove_index(index=index, model=Article)