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
|
import decimal
from django.db import NotSupportedError, connection
from django.db.backends.base.operations import BaseDatabaseOperations
from django.db.models import DurationField
from django.test import (
SimpleTestCase, TestCase, override_settings, skipIfDBFeature,
)
from django.utils import timezone
class SimpleDatabaseOperationTests(SimpleTestCase):
may_requre_msg = 'subclasses of BaseDatabaseOperations may require a %s() method'
def setUp(self):
self.ops = BaseDatabaseOperations(connection=connection)
def test_deferrable_sql(self):
self.assertEqual(self.ops.deferrable_sql(), '')
def test_end_transaction_rollback(self):
self.assertEqual(self.ops.end_transaction_sql(success=False), 'ROLLBACK;')
def test_no_limit_value(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'no_limit_value'):
self.ops.no_limit_value()
def test_quote_name(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'quote_name'):
self.ops.quote_name('a')
def test_regex_lookup(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'regex_lookup'):
self.ops.regex_lookup(lookup_type='regex')
def test_set_time_zone_sql(self):
self.assertEqual(self.ops.set_time_zone_sql(), '')
def test_sql_flush(self):
msg = 'subclasses of BaseDatabaseOperations must provide an sql_flush() method'
with self.assertRaisesMessage(NotImplementedError, msg):
self.ops.sql_flush(None, None, None)
def test_pk_default_value(self):
self.assertEqual(self.ops.pk_default_value(), 'DEFAULT')
def test_tablespace_sql(self):
self.assertEqual(self.ops.tablespace_sql(None), '')
def test_sequence_reset_by_name_sql(self):
self.assertEqual(self.ops.sequence_reset_by_name_sql(None, []), [])
def test_adapt_unknown_value_decimal(self):
value = decimal.Decimal('3.14')
self.assertEqual(
self.ops.adapt_unknown_value(value),
self.ops.adapt_decimalfield_value(value)
)
def test_adapt_unknown_value_date(self):
value = timezone.now().date()
self.assertEqual(self.ops.adapt_unknown_value(value), self.ops.adapt_datefield_value(value))
def test_adapt_unknown_value_time(self):
value = timezone.now().time()
self.assertEqual(self.ops.adapt_unknown_value(value), self.ops.adapt_timefield_value(value))
def test_adapt_timefield_value_none(self):
self.assertIsNone(self.ops.adapt_timefield_value(None))
def test_adapt_datetimefield_value(self):
self.assertIsNone(self.ops.adapt_datetimefield_value(None))
def test_adapt_timefield_value(self):
msg = 'Django does not support timezone-aware times.'
with self.assertRaisesMessage(ValueError, msg):
self.ops.adapt_timefield_value(timezone.make_aware(timezone.now()))
@override_settings(USE_TZ=False)
def test_adapt_timefield_value_unaware(self):
now = timezone.now()
self.assertEqual(self.ops.adapt_timefield_value(now), str(now))
def test_date_extract_sql(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'date_extract_sql'):
self.ops.date_extract_sql(None, None)
def test_time_extract_sql(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'date_extract_sql'):
self.ops.time_extract_sql(None, None)
def test_date_interval_sql(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'date_interval_sql'):
self.ops.date_interval_sql(None)
def test_date_trunc_sql(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'date_trunc_sql'):
self.ops.date_trunc_sql(None, None)
def test_time_trunc_sql(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'time_trunc_sql'):
self.ops.time_trunc_sql(None, None)
def test_datetime_trunc_sql(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'datetime_trunc_sql'):
self.ops.datetime_trunc_sql(None, None, None)
def test_datetime_cast_date_sql(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'datetime_cast_date_sql'):
self.ops.datetime_cast_date_sql(None, None)
def test_datetime_cast_time_sql(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'datetime_cast_time_sql'):
self.ops.datetime_cast_time_sql(None, None)
def test_datetime_extract_sql(self):
with self.assertRaisesMessage(NotImplementedError, self.may_requre_msg % 'datetime_extract_sql'):
self.ops.datetime_extract_sql(None, None, None)
class DatabaseOperationTests(TestCase):
def setUp(self):
self.ops = BaseDatabaseOperations(connection=connection)
@skipIfDBFeature('supports_over_clause')
def test_window_frame_raise_not_supported_error(self):
msg = 'This backend does not support window expressions.'
with self.assertRaisesMessage(NotSupportedError, msg):
self.ops.window_frame_rows_start_end()
@skipIfDBFeature('can_distinct_on_fields')
def test_distinct_on_fields(self):
msg = 'DISTINCT ON fields is not supported by this database backend'
with self.assertRaisesMessage(NotSupportedError, msg):
self.ops.distinct_sql(['a', 'b'], None)
@skipIfDBFeature('supports_temporal_subtraction')
def test_subtract_temporals(self):
duration_field = DurationField()
duration_field_internal_type = duration_field.get_internal_type()
msg = (
'This backend does not support %s subtraction.' %
duration_field_internal_type
)
with self.assertRaisesMessage(NotSupportedError, msg):
self.ops.subtract_temporals(duration_field_internal_type, None, None)
|