summaryrefslogtreecommitdiff
path: root/tests/queries/test_db_returning.py
blob: af9d041393f26173be50aa9a283177c2336bdf07 (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
import datetime

from django.db import NotSupportedError, connection
from django.test import TestCase, skipIfDBFeature, skipUnlessDBFeature
from django.test.utils import CaptureQueriesContext

from .models import DumbCategory, NonIntegerPKReturningModel, ReturningModel


@skipUnlessDBFeature('can_return_columns_from_insert')
class ReturningValuesTests(TestCase):
    def test_insert_returning(self):
        with CaptureQueriesContext(connection) as captured_queries:
            DumbCategory.objects.create()
        self.assertIn(
            'RETURNING %s.%s' % (
                connection.ops.quote_name(DumbCategory._meta.db_table),
                connection.ops.quote_name(DumbCategory._meta.get_field('id').column),
            ),
            captured_queries[-1]['sql'],
        )

    def test_insert_returning_non_integer(self):
        obj = NonIntegerPKReturningModel.objects.create()
        self.assertTrue(obj.created)
        self.assertIsInstance(obj.created, datetime.datetime)

    @skipUnlessDBFeature('can_return_multiple_columns_from_insert')
    def test_insert_returning_multiple(self):
        with CaptureQueriesContext(connection) as captured_queries:
            obj = ReturningModel.objects.create()
        table_name = connection.ops.quote_name(ReturningModel._meta.db_table)
        self.assertIn(
            'RETURNING %s.%s, %s.%s' % (
                table_name,
                connection.ops.quote_name(ReturningModel._meta.get_field('id').column),
                table_name,
                connection.ops.quote_name(ReturningModel._meta.get_field('created').column),
            ),
            captured_queries[-1]['sql'],
        )
        self.assertTrue(obj.pk)
        self.assertIsInstance(obj.created, datetime.datetime)

    @skipIfDBFeature('can_return_multiple_columns_from_insert')
    def test_insert_returning_multiple_not_supported(self):
        msg = (
            'Returning multiple columns from INSERT statements is '
            'not supported on this database backend.'
        )
        with self.assertRaisesMessage(NotSupportedError, msg):
            ReturningModel.objects.create()

    @skipUnlessDBFeature(
        'can_return_rows_from_bulk_insert',
        'can_return_multiple_columns_from_insert',
    )
    def test_bulk_insert(self):
        objs = [ReturningModel(), ReturningModel(pk=2 ** 11), ReturningModel()]
        ReturningModel.objects.bulk_create(objs)
        for obj in objs:
            with self.subTest(obj=obj):
                self.assertTrue(obj.pk)
                self.assertIsInstance(obj.created, datetime.datetime)