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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
|
import copy
import unittest
from unittest import mock
from django.core.exceptions import ImproperlyConfigured
from django.db import DatabaseError, NotSupportedError, ProgrammingError, connection
from django.db.models import BooleanField
from django.test import TestCase, TransactionTestCase
from ..models import Square, VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ
try:
from django.db.backends.oracle.oracledb_any import is_oracledb
except ImportError:
is_oracledb = False
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)
@unittest.skipUnless(is_oracledb, "Pool specific tests")
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()
@unittest.skipUnless(is_oracledb, "Pool specific tests")
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()
@unittest.skipUnless(is_oracledb, "Pool specific tests")
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()
@unittest.skipUnless(is_oracledb, "Pool specific tests")
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.skipIf(is_oracledb, "cx_oracle specific tests")
def test_cx_Oracle_not_support_pooling(self):
new_connection = no_pool_connection()
new_connection.settings_dict["OPTIONS"]["pool"] = True
msg = "Pooling isn't supported by cx_Oracle. Use python-oracledb instead."
with self.assertRaisesMessage(ImproperlyConfigured, msg):
new_connection.connect()
@unittest.skipUnless(connection.vendor == "oracle", "Oracle tests")
class TransactionalTests(TransactionTestCase):
available_apps = ["backends"]
def test_hidden_no_data_found_exception(self):
# "ORA-1403: no data found" exception is hidden by Oracle OCI library
# when an INSERT statement is used with a RETURNING clause (see #28859).
with connection.cursor() as cursor:
# Create trigger that raises "ORA-1403: no data found".
cursor.execute(
"""
CREATE OR REPLACE TRIGGER "TRG_NO_DATA_FOUND"
AFTER INSERT ON "BACKENDS_SQUARE"
FOR EACH ROW
BEGIN
RAISE NO_DATA_FOUND;
END;
"""
)
try:
with self.assertRaisesMessage(
DatabaseError,
(
'The database did not return a new row id. Probably "ORA-1403: no '
'data found" was raised internally but was hidden by the Oracle '
"OCI library (see https://code.djangoproject.com/ticket/28859)."
),
):
Square.objects.create(root=2, square=4)
finally:
with connection.cursor() as cursor:
cursor.execute('DROP TRIGGER "TRG_NO_DATA_FOUND"')
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
|
