Viewing github.com/django/django-stable-4.1.x/tests/backends/base/test

```
import decimal
```

from django.core.management.color import no_style

from django.db import NotSupportedError, connection, transaction

from django.db.backends.base.operations import BaseDatabaseOperations

from django.db.models import DurationField, Value

```
from django.test import (
```
```
    SimpleTestCase,
```
```
    TestCase,
```
```
    TransactionTestCase,
```
```
    override_settings,
```
```
    skipIfDBFeature,
```
```
)
```
```
from django.utils import timezone
```
```
from ..models import Author, Book
```

class SimpleDatabaseOperationTests(SimpleTestCase):

    may_require_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_require_msg % "no_limit_value"

```
        ):
```
```
            self.ops.no_limit_value()
```
```
    def test_quote_name(self):
```
```
        with self.assertRaisesMessage(
```

            NotImplementedError, self.may_require_msg % "quote_name"

```
        ):
```
```
            self.ops.quote_name("a")
```
```
    def test_regex_lookup(self):
```
```
        with self.assertRaisesMessage(
```

            NotImplementedError, self.may_require_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)

```
    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_timefield_value_expression(self):

        value = Value(timezone.now().time())

        self.assertEqual(self.ops.adapt_timefield_value(value), value)

    def test_adapt_datetimefield_value_none(self):

        self.assertIsNone(self.ops.adapt_datetimefield_value(None))

    def test_adapt_datetimefield_value_expression(self):

```
        value = Value(timezone.now())
```

        self.assertEqual(self.ops.adapt_datetimefield_value(value), value)

    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_format_for_duration_arithmetic(self):

        msg = self.may_require_msg % "format_for_duration_arithmetic"

        with self.assertRaisesMessage(NotImplementedError, msg):

            self.ops.format_for_duration_arithmetic(None)

```
    def test_date_extract_sql(self):
```
```
        with self.assertRaisesMessage(
```

            NotImplementedError, self.may_require_msg % "date_extract_sql"

```
        ):
```

            self.ops.date_extract_sql(None, None, None)

```
    def test_time_extract_sql(self):
```
```
        with self.assertRaisesMessage(
```

            NotImplementedError, self.may_require_msg % "date_extract_sql"

```
        ):
```

            self.ops.time_extract_sql(None, None, None)

```
    def test_date_trunc_sql(self):
```
```
        with self.assertRaisesMessage(
```

            NotImplementedError, self.may_require_msg % "date_trunc_sql"

```
        ):
```

            self.ops.date_trunc_sql(None, None, None)

```
    def test_time_trunc_sql(self):
```
```
        with self.assertRaisesMessage(
```

            NotImplementedError, self.may_require_msg % "time_trunc_sql"

```
        ):
```

            self.ops.time_trunc_sql(None, None, None)

```
    def test_datetime_trunc_sql(self):
```
```
        with self.assertRaisesMessage(
```

            NotImplementedError, self.may_require_msg % "datetime_trunc_sql"

```
        ):
```

            self.ops.datetime_trunc_sql(None, None, None, None)

    def test_datetime_cast_date_sql(self):

```
        with self.assertRaisesMessage(
```

            NotImplementedError, self.may_require_msg % "datetime_cast_date_sql"

```
        ):
```

            self.ops.datetime_cast_date_sql(None, None, None)

    def test_datetime_cast_time_sql(self):

```
        with self.assertRaisesMessage(
```

            NotImplementedError, self.may_require_msg % "datetime_cast_time_sql"

```
        ):
```

            self.ops.datetime_cast_time_sql(None, None, None)

    def test_datetime_extract_sql(self):

```
        with self.assertRaisesMessage(
```

            NotImplementedError, self.may_require_msg % "datetime_extract_sql"

```
        ):
```

            self.ops.datetime_extract_sql(None, 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)

class SqlFlushTests(TransactionTestCase):

```
    available_apps = ["backends"]
```

    def test_sql_flush_no_tables(self):

        self.assertEqual(connection.ops.sql_flush(no_style(), []), [])

    def test_execute_sql_flush_statements(self):

```
        with transaction.atomic():
```

            author = Author.objects.create(name="George Orwell")

            Book.objects.create(author=author)

            author = Author.objects.create(name="Harper Lee")

            Book.objects.create(author=author)

            Book.objects.create(author=author)

            self.assertIs(Author.objects.exists(), True)

            self.assertIs(Book.objects.exists(), True)

        sql_list = connection.ops.sql_flush(

```
            no_style(),
```

            [Author._meta.db_table, Book._meta.db_table],

```
            reset_sequences=True,
```
```
            allow_cascade=True,
```
```
        )
```

        connection.ops.execute_sql_flush(sql_list)

```
        with transaction.atomic():
```

            self.assertIs(Author.objects.exists(), False)

            self.assertIs(Book.objects.exists(), False)

            if connection.features.supports_sequence_reset:

                author = Author.objects.create(name="F. Scott Fitzgerald")

                self.assertEqual(author.pk, 1)

                book = Book.objects.create(author=author)

                self.assertEqual(book.pk, 1)