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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
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
|
