summaryrefslogtreecommitdiff
path: root/tests/indexes/tests.py
blob: 219dfe67b147e54f4598dab9559be9a1406ad5e0 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
from unittest import skipIf, skipUnless

from django.db import connection
from django.db.models import Index
from django.db.models.deletion import CASCADE
from django.db.models.fields.related import ForeignKey
from django.test import TestCase, TransactionTestCase

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.
        """
        with connection.schema_editor() as 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)
        with connection.schema_editor() as 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)


@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 ingored.
            editor.add_index(IndexedArticle2, index)


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


@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).
        """
        storage = connection.introspection.get_storage_engine(
            connection.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
                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`)'
                ])
        finally:
            if field_created:
                with connection.schema_editor() as editor:
                    editor.remove_field(ArticleTranslation, new_field)