django/tests/backends/oracle/tests.py

import copy
import unittest
from unittest import mock

from django.core.exceptions import ImproperlyConfigured
from django.db import NotSupportedError, ProgrammingError, connection
from django.db.models import BooleanField
from django.test import TestCase, TransactionTestCase

from ..models import VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ


def no_pool_connection(alias=None):
    new_connection = connection.copy(alias)
    new_connection.settings_dict = copy.deepcopy(connection.settings_dict)
    # Ensure that the second connection circumvents the pool, this is kind
    # of a hack, but we cannot easily change the pool connections.
    new_connection.settings_dict["OPTIONS"]["pool"] = False
    return new_connection


@unittest.skipUnless(connection.vendor == "oracle", "Oracle tests")
class Tests(TestCase):
    def test_quote_name(self):
        """'%' chars are escaped for query execution."""
        name = '"SOME%NAME"'
        quoted_name = connection.ops.quote_name(name)
        self.assertEqual(quoted_name % (), name)

    def test_quote_name_db_table(self):
        model = VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ
        db_table = model._meta.db_table.upper()
        self.assertEqual(
            f'"{db_table}"',
            connection.ops.quote_name(
                "backends_verylongmodelnamezzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz",
            ),
        )

    def test_dbms_session(self):
        """A stored procedure can be called through a cursor wrapper."""
        with connection.cursor() as cursor:
            cursor.callproc("DBMS_SESSION.SET_IDENTIFIER", ["_django_testing!"])

    def test_cursor_var(self):
        """Cursor variables can be passed as query parameters."""
        with connection.cursor() as cursor:
            var = cursor.var(str)
            cursor.execute("BEGIN %s := 'X'; END; ", [var])
            self.assertEqual(var.getvalue(), "X")

    def test_order_of_nls_parameters(self):
        """
        An 'almost right' datetime works with configured NLS parameters
        (#18465).
        """
        suffix = connection.features.bare_select_suffix
        with connection.cursor() as cursor:
            query = f"SELECT 1{suffix} WHERE '1936-12-29 00:00' < SYSDATE"
            # The query succeeds without errors - pre #18465 this
            # wasn't the case.
            cursor.execute(query)
            self.assertEqual(cursor.fetchone()[0], 1)

    def test_boolean_constraints(self):
        """Boolean fields have check constraints on their values."""
        for field in (BooleanField(), BooleanField(null=True)):
            with self.subTest(field=field):
                field.set_attributes_from_name("is_nice")
                self.assertIn('"IS_NICE" IN (0,1)', field.db_check(connection))

    @mock.patch.object(
        connection,
        "get_database_version",
        return_value=(18, 1),
    )
    def test_check_database_version_supported(self, mocked_get_database_version):
        msg = "Oracle 19 or later is required (found 18.1)."
        with self.assertRaisesMessage(NotSupportedError, msg):
            connection.check_database_version_supported()
        self.assertTrue(mocked_get_database_version.called)

    def test_pool_set_to_true(self):
        new_connection = no_pool_connection(alias="default_pool")
        new_connection.settings_dict["OPTIONS"]["pool"] = True
        try:
            self.assertIsNotNone(new_connection.pool)
        finally:
            new_connection.close_pool()

    def test_pool_reuse(self):
        new_connection = no_pool_connection(alias="default_pool")
        new_connection.settings_dict["OPTIONS"]["pool"] = {
            "min": 0,
            "max": 2,
        }
        self.assertIsNotNone(new_connection.pool)

        connections = []

        def get_connection():
            # copy() reuses the existing alias and as such the same pool.
            conn = new_connection.copy()
            conn.connect()
            connections.append(conn)
            return conn

        try:
            connection_1 = get_connection()  # First connection.
            get_connection()  # Get the second connection.
            sql = "select sys_context('userenv', 'sid') from dual"
            sids = [conn.cursor().execute(sql).fetchone()[0] for conn in connections]
            connection_1.close()  # Release back to the pool.
            connection_3 = get_connection()
            sid = connection_3.cursor().execute(sql).fetchone()[0]
            # Reuses the first connection as it is available.
            self.assertEqual(sid, sids[0])
        finally:
            # Release all connections back to the pool.
            for conn in connections:
                conn.close()
            new_connection.close_pool()

    def test_cannot_open_new_connection_in_atomic_block(self):
        new_connection = no_pool_connection(alias="default_pool")
        new_connection.settings_dict["OPTIONS"]["pool"] = True
        msg = "Cannot open a new connection in an atomic block."
        new_connection.in_atomic_block = True
        new_connection.closed_in_transaction = True
        with self.assertRaisesMessage(ProgrammingError, msg):
            new_connection.ensure_connection()

    def test_pooling_not_support_persistent_connections(self):
        new_connection = no_pool_connection(alias="default_pool")
        new_connection.settings_dict["OPTIONS"]["pool"] = True
        new_connection.settings_dict["CONN_MAX_AGE"] = 10
        msg = "Pooling doesn't support persistent connections."
        with self.assertRaisesMessage(ImproperlyConfigured, msg):
            new_connection.pool


@unittest.skipUnless(connection.vendor == "oracle", "Oracle tests")
class TransactionalTests(TransactionTestCase):
    available_apps = ["backends"]

    def test_password_with_at_sign(self):
        from django.db.backends.oracle.base import Database

        old_password = connection.settings_dict["PASSWORD"]
        connection.settings_dict["PASSWORD"] = "p@ssword"
        try:
            self.assertIn(
                '/"p@ssword"@',
                connection.client.connect_string(connection.settings_dict),
            )
            with self.assertRaises(Database.DatabaseError) as context:
                connection.connect()
            # Database exception: "ORA-01017: invalid username/password" is
            # expected.
            self.assertIn("ORA-01017", context.exception.args[0].message)
        finally:
            connection.settings_dict["PASSWORD"] = old_password