Tests indexes

models.py

 1from django.db import connection
 2from django.db import models
 3
 4
 5class CurrentTranslation(models.ForeignObject):
 6    """
 7    Creates virtual relation to the translation with model cache enabled.
 8    """
 9
10    # Avoid validation
11    requires_unique_target = False
12
13    def __init__(self, to, on_delete, from_fields, to_fields, **kwargs):
14        # Disable reverse relation
15        kwargs["related_name"] = "+"
16        # Set unique to enable model cache.
17        kwargs["unique"] = True
18        super().__init__(to, on_delete, from_fields, to_fields, **kwargs)
19
20
21class ArticleTranslation(models.Model):
22
23    article = models.ForeignKey("indexes.Article", models.CASCADE)
24    article_no_constraint = models.ForeignKey(
25        "indexes.Article", models.CASCADE, db_constraint=False, related_name="+"
26    )
27    language = models.CharField(max_length=10, unique=True)
28    content = models.TextField()
29
30
31class Article(models.Model):
32    headline = models.CharField(max_length=100)
33    pub_date = models.DateTimeField()
34    published = models.BooleanField(default=False)
35
36    # Add virtual relation to the ArticleTranslation model.
37    translation = CurrentTranslation(
38        ArticleTranslation, models.CASCADE, ["id"], ["article"]
39    )
40
41    class Meta:
42        index_together = [
43            ["headline", "pub_date"],
44        ]
45
46
47# Model for index_together being used only with single list
48class IndexTogetherSingleList(models.Model):
49    headline = models.CharField(max_length=100)
50    pub_date = models.DateTimeField()
51
52    class Meta:
53        index_together = ["headline", "pub_date"]
54
55
56# Indexing a TextField on Oracle or MySQL results in index creation error.
57if connection.vendor == "postgresql":
58
59    class IndexedArticle(models.Model):
60        headline = models.CharField(max_length=100, db_index=True)
61        body = models.TextField(db_index=True)
62        slug = models.CharField(max_length=40, unique=True)
63
64
65class IndexedArticle2(models.Model):
66    headline = models.CharField(max_length=100)
67    body = models.TextField()

tests.py

  1import datetime
  2from unittest import skipIf
  3from unittest import skipUnless
  4
  5from django.db import connection
  6from django.db.models import CASCADE
  7from django.db.models import ForeignKey
  8from django.db.models import Index
  9from django.db.models import Q
 10from django.test import skipIfDBFeature
 11from django.test import skipUnlessDBFeature
 12from django.test import TestCase
 13from django.test import TransactionTestCase
 14from django.test.utils import override_settings
 15from django.utils import timezone
 16
 17from .models import Article
 18from .models import ArticleTranslation
 19from .models import IndexedArticle2
 20from .models import IndexTogetherSingleList
 21
 22
 23class SchemaIndexesTests(TestCase):
 24    """
 25    Test index handling by the db.backends.schema infrastructure.
 26    """
 27
 28    def test_index_name_hash(self):
 29        """
 30        Index names should be deterministic.
 31        """
 32        editor = connection.schema_editor()
 33        index_name = editor._create_index_name(
 34            table_name=Article._meta.db_table,
 35            column_names=("c1",),
 36            suffix="123",
 37        )
 38        self.assertEqual(index_name, "indexes_article_c1_a52bd80b123")
 39
 40    def test_index_name(self):
 41        """
 42        Index names on the built-in database backends::
 43            * Are truncated as needed.
 44            * Include all the column names.
 45            * Include a deterministic hash.
 46        """
 47        long_name = "l%sng" % ("o" * 100)
 48        editor = connection.schema_editor()
 49        index_name = editor._create_index_name(
 50            table_name=Article._meta.db_table,
 51            column_names=("c1", "c2", long_name),
 52            suffix="ix",
 53        )
 54        expected = {
 55            "mysql": "indexes_article_c1_c2_looooooooooooooooooo_255179b2ix",
 56            "oracle": "indexes_a_c1_c2_loo_255179b2ix",
 57            "postgresql": "indexes_article_c1_c2_loooooooooooooooooo_255179b2ix",
 58            "sqlite": "indexes_article_c1_c2_l%sng_255179b2ix" % ("o" * 100),
 59        }
 60        if connection.vendor not in expected:
 61            self.skipTest(
 62                "This test is only supported on the built-in database backends."
 63            )
 64        self.assertEqual(index_name, expected[connection.vendor])
 65
 66    def test_index_together(self):
 67        editor = connection.schema_editor()
 68        index_sql = [str(statement) for statement in editor._model_indexes_sql(Article)]
 69        self.assertEqual(len(index_sql), 1)
 70        # Ensure the index name is properly quoted
 71        self.assertIn(
 72            connection.ops.quote_name(
 73                editor._create_index_name(
 74                    Article._meta.db_table, ["headline", "pub_date"], suffix="_idx"
 75                )
 76            ),
 77            index_sql[0],
 78        )
 79
 80    def test_index_together_single_list(self):
 81        # Test for using index_together with a single list (#22172)
 82        index_sql = connection.schema_editor()._model_indexes_sql(
 83            IndexTogetherSingleList
 84        )
 85        self.assertEqual(len(index_sql), 1)
 86
 87    def test_columns_list_sql(self):
 88        index = Index(fields=["headline"], name="whitespace_idx")
 89        editor = connection.schema_editor()
 90        self.assertIn(
 91            "(%s)" % editor.quote_name("headline"),
 92            str(index.create_sql(Article, editor)),
 93        )
 94
 95    def test_descending_columns_list_sql(self):
 96        index = Index(fields=["-headline"], name="whitespace_idx")
 97        editor = connection.schema_editor()
 98        self.assertIn(
 99            "(%s DESC)" % editor.quote_name("headline"),
100            str(index.create_sql(Article, editor)),
101        )
102
103
104@skipIf(connection.vendor == "postgresql", "opclasses are PostgreSQL only")
105class SchemaIndexesNotPostgreSQLTests(TransactionTestCase):
106    available_apps = ["indexes"]
107
108    def test_create_index_ignores_opclasses(self):
109        index = Index(
110            name="test_ops_class",
111            fields=["headline"],
112            opclasses=["varchar_pattern_ops"],
113        )
114        with connection.schema_editor() as editor:
115            # This would error if opclasses weren't ignored.
116            editor.add_index(IndexedArticle2, index)
117
118
119# The `condition` parameter is ignored by databases that don't support partial
120# indexes.
121@skipIfDBFeature("supports_partial_indexes")
122class PartialIndexConditionIgnoredTests(TransactionTestCase):
123    available_apps = ["indexes"]
124
125    def test_condition_ignored(self):
126        index = Index(
127            name="test_condition_ignored",
128            fields=["published"],
129            condition=Q(published=True),
130        )
131        with connection.schema_editor() as editor:
132            # This would error if condition weren't ignored.
133            editor.add_index(Article, index)
134
135        self.assertNotIn(
136            "WHERE %s" % editor.quote_name("published"),
137            str(index.create_sql(Article, editor)),
138        )
139
140
141@skipUnless(connection.vendor == "postgresql", "PostgreSQL tests")
142class SchemaIndexesPostgreSQLTests(TransactionTestCase):
143    available_apps = ["indexes"]
144    get_opclass_query = """
145        SELECT opcname, c.relname FROM pg_opclass AS oc
146        JOIN pg_index as i on oc.oid = ANY(i.indclass)
147        JOIN pg_class as c on c.oid = i.indexrelid
148        WHERE c.relname = '%s'
149    """
150
151    def test_text_indexes(self):
152        """Test creation of PostgreSQL-specific text indexes (#12234)"""
153        from .models import IndexedArticle
154
155        index_sql = [
156            str(statement)
157            for statement in connection.schema_editor()._model_indexes_sql(
158                IndexedArticle
159            )
160        ]
161        self.assertEqual(len(index_sql), 5)
162        self.assertIn('("headline" varchar_pattern_ops)', index_sql[1])
163        self.assertIn('("body" text_pattern_ops)', index_sql[3])
164        # unique=True and db_index=True should only create the varchar-specific
165        # index (#19441).
166        self.assertIn('("slug" varchar_pattern_ops)', index_sql[4])
167
168    def test_virtual_relation_indexes(self):
169        """Test indexes are not created for related objects"""
170        index_sql = connection.schema_editor()._model_indexes_sql(Article)
171        self.assertEqual(len(index_sql), 1)
172
173    def test_ops_class(self):
174        index = Index(
175            name="test_ops_class",
176            fields=["headline"],
177            opclasses=["varchar_pattern_ops"],
178        )
179        with connection.schema_editor() as editor:
180            editor.add_index(IndexedArticle2, index)
181        with editor.connection.cursor() as cursor:
182            cursor.execute(self.get_opclass_query % "test_ops_class")
183            self.assertEqual(
184                cursor.fetchall(), [("varchar_pattern_ops", "test_ops_class")]
185            )
186
187    def test_ops_class_multiple_columns(self):
188        index = Index(
189            name="test_ops_class_multiple",
190            fields=["headline", "body"],
191            opclasses=["varchar_pattern_ops", "text_pattern_ops"],
192        )
193        with connection.schema_editor() as editor:
194            editor.add_index(IndexedArticle2, index)
195        with editor.connection.cursor() as cursor:
196            cursor.execute(self.get_opclass_query % "test_ops_class_multiple")
197            expected_ops_classes = (
198                ("varchar_pattern_ops", "test_ops_class_multiple"),
199                ("text_pattern_ops", "test_ops_class_multiple"),
200            )
201            self.assertCountEqual(cursor.fetchall(), expected_ops_classes)
202
203    def test_ops_class_partial(self):
204        index = Index(
205            name="test_ops_class_partial",
206            fields=["body"],
207            opclasses=["text_pattern_ops"],
208            condition=Q(headline__contains="China"),
209        )
210        with connection.schema_editor() as editor:
211            editor.add_index(IndexedArticle2, index)
212        with editor.connection.cursor() as cursor:
213            cursor.execute(self.get_opclass_query % "test_ops_class_partial")
214            self.assertCountEqual(
215                cursor.fetchall(), [("text_pattern_ops", "test_ops_class_partial")]
216            )
217
218    def test_ops_class_partial_tablespace(self):
219        indexname = "test_ops_class_tblspace"
220        index = Index(
221            name=indexname,
222            fields=["body"],
223            opclasses=["text_pattern_ops"],
224            condition=Q(headline__contains="China"),
225            db_tablespace="pg_default",
226        )
227        with connection.schema_editor() as editor:
228            editor.add_index(IndexedArticle2, index)
229            self.assertIn(
230                'TABLESPACE "pg_default" ',
231                str(index.create_sql(IndexedArticle2, editor)),
232            )
233        with editor.connection.cursor() as cursor:
234            cursor.execute(self.get_opclass_query % indexname)
235            self.assertCountEqual(cursor.fetchall(), [("text_pattern_ops", indexname)])
236
237    def test_ops_class_descending(self):
238        indexname = "test_ops_class_ordered"
239        index = Index(
240            name=indexname,
241            fields=["-body"],
242            opclasses=["text_pattern_ops"],
243        )
244        with connection.schema_editor() as editor:
245            editor.add_index(IndexedArticle2, index)
246        with editor.connection.cursor() as cursor:
247            cursor.execute(self.get_opclass_query % indexname)
248            self.assertCountEqual(cursor.fetchall(), [("text_pattern_ops", indexname)])
249
250    def test_ops_class_descending_partial(self):
251        indexname = "test_ops_class_ordered_partial"
252        index = Index(
253            name=indexname,
254            fields=["-body"],
255            opclasses=["text_pattern_ops"],
256            condition=Q(headline__contains="China"),
257        )
258        with connection.schema_editor() as editor:
259            editor.add_index(IndexedArticle2, index)
260        with editor.connection.cursor() as cursor:
261            cursor.execute(self.get_opclass_query % indexname)
262            self.assertCountEqual(cursor.fetchall(), [("text_pattern_ops", indexname)])
263
264    def test_ops_class_columns_lists_sql(self):
265        index = Index(
266            fields=["headline"],
267            name="whitespace_idx",
268            opclasses=["text_pattern_ops"],
269        )
270        with connection.schema_editor() as editor:
271            self.assertIn(
272                "(%s text_pattern_ops)" % editor.quote_name("headline"),
273                str(index.create_sql(Article, editor)),
274            )
275
276    def test_ops_class_descending_columns_list_sql(self):
277        index = Index(
278            fields=["-headline"],
279            name="whitespace_idx",
280            opclasses=["text_pattern_ops"],
281        )
282        with connection.schema_editor() as editor:
283            self.assertIn(
284                "(%s text_pattern_ops DESC)" % editor.quote_name("headline"),
285                str(index.create_sql(Article, editor)),
286            )
287
288
289@skipUnless(connection.vendor == "mysql", "MySQL tests")
290class SchemaIndexesMySQLTests(TransactionTestCase):
291    available_apps = ["indexes"]
292
293    def test_no_index_for_foreignkey(self):
294        """
295        MySQL on InnoDB already creates indexes automatically for foreign keys.
296        (#14180). An index should be created if db_constraint=False (#26171).
297        """
298        with connection.cursor() as cursor:
299            storage = connection.introspection.get_storage_engine(
300                cursor,
301                ArticleTranslation._meta.db_table,
302            )
303        if storage != "InnoDB":
304            self.skip("This test only applies to the InnoDB storage engine")
305        index_sql = [
306            str(statement)
307            for statement in connection.schema_editor()._model_indexes_sql(
308                ArticleTranslation
309            )
310        ]
311        self.assertEqual(
312            index_sql,
313            [
314                "CREATE INDEX `indexes_articletranslation_article_no_constraint_id_d6c0806b` "
315                "ON `indexes_articletranslation` (`article_no_constraint_id`)"
316            ],
317        )
318
319        # The index also shouldn't be created if the ForeignKey is added after
320        # the model was created.
321        field_created = False
322        try:
323            with connection.schema_editor() as editor:
324                new_field = ForeignKey(Article, CASCADE)
325                new_field.set_attributes_from_name("new_foreign_key")
326                editor.add_field(ArticleTranslation, new_field)
327                field_created = True
328                # No deferred SQL. The FK constraint is included in the
329                # statement to add the field.
330                self.assertFalse(editor.deferred_sql)
331        finally:
332            if field_created:
333                with connection.schema_editor() as editor:
334                    editor.remove_field(ArticleTranslation, new_field)
335
336
337@skipUnlessDBFeature("supports_partial_indexes")
338# SQLite doesn't support timezone-aware datetimes when USE_TZ is False.
339@override_settings(USE_TZ=True)
340class PartialIndexTests(TransactionTestCase):
341    # Schema editor is used to create the index to test that it works.
342    available_apps = ["indexes"]
343
344    def test_partial_index(self):
345        with connection.schema_editor() as editor:
346            index = Index(
347                name="recent_article_idx",
348                fields=["pub_date"],
349                condition=Q(
350                    pub_date__gt=datetime.datetime(
351                        year=2015,
352                        month=1,
353                        day=1,
354                        # PostgreSQL would otherwise complain about the lookup
355                        # being converted to a mutable function (by removing
356                        # the timezone in the cast) which is forbidden.
357                        tzinfo=timezone.get_current_timezone(),
358                    ),
359                ),
360            )
361            self.assertIn(
362                "WHERE %s" % editor.quote_name("pub_date"),
363                str(index.create_sql(Article, schema_editor=editor)),
364            )
365            editor.add_index(index=index, model=Article)
366            with connection.cursor() as cursor:
367                self.assertIn(
368                    index.name,
369                    connection.introspection.get_constraints(
370                        cursor=cursor,
371                        table_name=Article._meta.db_table,
372                    ),
373                )
374            editor.remove_index(index=index, model=Article)
375
376    def test_integer_restriction_partial(self):
377        with connection.schema_editor() as editor:
378            index = Index(
379                name="recent_article_idx",
380                fields=["id"],
381                condition=Q(pk__gt=1),
382            )
383            self.assertIn(
384                "WHERE %s" % editor.quote_name("id"),
385                str(index.create_sql(Article, schema_editor=editor)),
386            )
387            editor.add_index(index=index, model=Article)
388            with connection.cursor() as cursor:
389                self.assertIn(
390                    index.name,
391                    connection.introspection.get_constraints(
392                        cursor=cursor,
393                        table_name=Article._meta.db_table,
394                    ),
395                )
396            editor.remove_index(index=index, model=Article)
397
398    def test_boolean_restriction_partial(self):
399        with connection.schema_editor() as editor:
400            index = Index(
401                name="published_index",
402                fields=["published"],
403                condition=Q(published=True),
404            )
405            self.assertIn(
406                "WHERE %s" % editor.quote_name("published"),
407                str(index.create_sql(Article, schema_editor=editor)),
408            )
409            editor.add_index(index=index, model=Article)
410            with connection.cursor() as cursor:
411                self.assertIn(
412                    index.name,
413                    connection.introspection.get_constraints(
414                        cursor=cursor,
415                        table_name=Article._meta.db_table,
416                    ),
417                )
418            editor.remove_index(index=index, model=Article)
419
420    @skipUnlessDBFeature("supports_functions_in_partial_indexes")
421    def test_multiple_conditions(self):
422        with connection.schema_editor() as editor:
423            index = Index(
424                name="recent_article_idx",
425                fields=["pub_date", "headline"],
426                condition=(
427                    Q(
428                        pub_date__gt=datetime.datetime(
429                            year=2015,
430                            month=1,
431                            day=1,
432                            tzinfo=timezone.get_current_timezone(),
433                        )
434                    )
435                    & Q(headline__contains="China")
436                ),
437            )
438            sql = str(index.create_sql(Article, schema_editor=editor))
439            where = sql.find("WHERE")
440            self.assertIn("WHERE (%s" % editor.quote_name("pub_date"), sql)
441            # Because each backend has different syntax for the operators,
442            # check ONLY the occurrence of headline in the SQL.
443            self.assertGreater(sql.rfind("headline"), where)
444            editor.add_index(index=index, model=Article)
445            with connection.cursor() as cursor:
446                self.assertIn(
447                    index.name,
448                    connection.introspection.get_constraints(
449                        cursor=cursor,
450                        table_name=Article._meta.db_table,
451                    ),
452                )
453            editor.remove_index(index=index, model=Article)
454
455    def test_is_null_condition(self):
456        with connection.schema_editor() as editor:
457            index = Index(
458                name="recent_article_idx",
459                fields=["pub_date"],
460                condition=Q(pub_date__isnull=False),
461            )
462            self.assertIn(
463                "WHERE %s IS NOT NULL" % editor.quote_name("pub_date"),
464                str(index.create_sql(Article, schema_editor=editor)),
465            )
466            editor.add_index(index=index, model=Article)
467            with connection.cursor() as cursor:
468                self.assertIn(
469                    index.name,
470                    connection.introspection.get_constraints(
471                        cursor=cursor,
472                        table_name=Article._meta.db_table,
473                    ),
474                )
475            editor.remove_index(index=index, model=Article)