Refs #31233 -- Changed DatabaseWrapper._nodb_connection to _nodb_cursor().

It is now a method instead of a property and returns a context manager
that yields a cursor on entry and closes the cursor and connection upon
exit.
This commit is contained in:
Jon Dufresne 2020-02-03 19:07:00 -08:00 committed by Mariusz Felisiak
parent 2d55cb5c4a
commit f48f671223
8 changed files with 58 additions and 34 deletions

View file

@ -4,24 +4,32 @@ from unittest import mock
from django.core.exceptions import ImproperlyConfigured
from django.db import DatabaseError, connection, connections
from django.db.backends.base.base import BaseDatabaseWrapper
from django.test import TestCase, override_settings
@unittest.skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests')
class Tests(TestCase):
databases = {'default', 'other'}
def test_nodb_connection(self):
def test_nodb_cursor(self):
"""
The _nodb_connection property fallbacks to the default connection
database when access to the 'postgres' database is not granted.
The _nodb_cursor() fallbacks to the default connection database when
access to the 'postgres' database is not granted.
"""
orig_connect = BaseDatabaseWrapper.connect
def mocked_connect(self):
if self.settings_dict['NAME'] is None:
raise DatabaseError()
return ''
return orig_connect(self)
nodb_conn = connection._nodb_connection
self.assertIsNone(nodb_conn.settings_dict['NAME'])
with connection._nodb_cursor() as cursor:
self.assertIs(cursor.closed, False)
self.assertIsNotNone(cursor.db.connection)
self.assertIsNone(cursor.db.settings_dict['NAME'])
self.assertIs(cursor.closed, True)
self.assertIsNone(cursor.db.connection)
# Now assume the 'postgres' db isn't available
msg = (
@ -39,9 +47,13 @@ class Tests(TestCase):
'settings_dict',
{**connection.settings_dict, 'NAME': 'postgres'},
):
nodb_conn = connection._nodb_connection
self.assertIsNotNone(nodb_conn.settings_dict['NAME'])
self.assertEqual(nodb_conn.settings_dict['NAME'], connections['other'].settings_dict['NAME'])
with connection._nodb_cursor() as cursor:
self.assertIs(cursor.closed, False)
self.assertIsNotNone(cursor.db.connection)
self.assertIs(cursor.closed, True)
self.assertIsNone(cursor.db.connection)
self.assertIsNotNone(cursor.db.settings_dict['NAME'])
self.assertEqual(cursor.db.settings_dict['NAME'], connections['other'].settings_dict['NAME'])
def test_database_name_too_long(self):
from django.db.backends.postgresql.base import DatabaseWrapper