summaryrefslogtreecommitdiff
path: root/tests/backends/postgresql
diff options
context:
space:
mode:
Diffstat (limited to 'tests/backends/postgresql')
-rw-r--r--tests/backends/postgresql/__init__.py0
-rw-r--r--tests/backends/postgresql/test_creation.py95
-rw-r--r--tests/backends/postgresql/test_server_side_cursors.py82
-rw-r--r--tests/backends/postgresql/tests.py147
4 files changed, 324 insertions, 0 deletions
diff --git a/tests/backends/postgresql/__init__.py b/tests/backends/postgresql/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/tests/backends/postgresql/__init__.py
diff --git a/tests/backends/postgresql/test_creation.py b/tests/backends/postgresql/test_creation.py
new file mode 100644
index 0000000000..9554e97a54
--- /dev/null
+++ b/tests/backends/postgresql/test_creation.py
@@ -0,0 +1,95 @@
+import unittest
+from contextlib import contextmanager
+from io import StringIO
+from unittest import mock
+
+from django.db import connection
+from django.db.backends.base.creation import BaseDatabaseCreation
+from django.db.utils import DatabaseError
+from django.test import SimpleTestCase
+
+try:
+ import psycopg2 # NOQA
+except ImportError:
+ pass
+else:
+ from psycopg2 import errorcodes
+ from django.db.backends.postgresql.creation import DatabaseCreation
+
+
+@unittest.skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests')
+class DatabaseCreationTests(SimpleTestCase):
+
+ @contextmanager
+ def changed_test_settings(self, **kwargs):
+ settings = connection.settings_dict['TEST']
+ saved_values = {}
+ for name in kwargs:
+ if name in settings:
+ saved_values[name] = settings[name]
+
+ for name, value in kwargs.items():
+ settings[name] = value
+ try:
+ yield
+ finally:
+ for name, value in kwargs.items():
+ if name in saved_values:
+ settings[name] = saved_values[name]
+ else:
+ del settings[name]
+
+ def check_sql_table_creation_suffix(self, settings, expected):
+ with self.changed_test_settings(**settings):
+ creation = DatabaseCreation(connection)
+ suffix = creation.sql_table_creation_suffix()
+ self.assertEqual(suffix, expected)
+
+ def test_sql_table_creation_suffix_with_none_settings(self):
+ settings = {'CHARSET': None, 'TEMPLATE': None}
+ self.check_sql_table_creation_suffix(settings, "")
+
+ def test_sql_table_creation_suffix_with_encoding(self):
+ settings = {'CHARSET': 'UTF8'}
+ self.check_sql_table_creation_suffix(settings, "WITH ENCODING 'UTF8'")
+
+ def test_sql_table_creation_suffix_with_template(self):
+ settings = {'TEMPLATE': 'template0'}
+ self.check_sql_table_creation_suffix(settings, 'WITH TEMPLATE "template0"')
+
+ def test_sql_table_creation_suffix_with_encoding_and_template(self):
+ settings = {'CHARSET': 'UTF8', 'TEMPLATE': 'template0'}
+ self.check_sql_table_creation_suffix(settings, '''WITH ENCODING 'UTF8' TEMPLATE "template0"''')
+
+ def _execute_raise_database_already_exists(self, cursor, parameters, keepdb=False):
+ error = DatabaseError('database %s already exists' % parameters['dbname'])
+ error.pgcode = errorcodes.DUPLICATE_DATABASE
+ raise DatabaseError() from error
+
+ def _execute_raise_permission_denied(self, cursor, parameters, keepdb=False):
+ error = DatabaseError('permission denied to create database')
+ error.pgcode = errorcodes.INSUFFICIENT_PRIVILEGE
+ raise DatabaseError() from error
+
+ def patch_test_db_creation(self, execute_create_test_db):
+ return mock.patch.object(BaseDatabaseCreation, '_execute_create_test_db', execute_create_test_db)
+
+ @mock.patch('sys.stdout', new_callable=StringIO)
+ @mock.patch('sys.stderr', new_callable=StringIO)
+ def test_create_test_db(self, *mocked_objects):
+ creation = DatabaseCreation(connection)
+ # Simulate test database creation raising "database already exists"
+ with self.patch_test_db_creation(self._execute_raise_database_already_exists):
+ with mock.patch('builtins.input', return_value='no'):
+ with self.assertRaises(SystemExit):
+ # SystemExit is raised if the user answers "no" to the
+ # prompt asking if it's okay to delete the test database.
+ creation._create_test_db(verbosity=0, autoclobber=False, keepdb=False)
+ # "Database already exists" error is ignored when keepdb is on
+ creation._create_test_db(verbosity=0, autoclobber=False, keepdb=True)
+ # Simulate test database creation raising unexpected error
+ with self.patch_test_db_creation(self._execute_raise_permission_denied):
+ with self.assertRaises(SystemExit):
+ creation._create_test_db(verbosity=0, autoclobber=False, keepdb=False)
+ with self.assertRaises(SystemExit):
+ creation._create_test_db(verbosity=0, autoclobber=False, keepdb=True)
diff --git a/tests/backends/postgresql/test_server_side_cursors.py b/tests/backends/postgresql/test_server_side_cursors.py
new file mode 100644
index 0000000000..8576686211
--- /dev/null
+++ b/tests/backends/postgresql/test_server_side_cursors.py
@@ -0,0 +1,82 @@
+import operator
+import unittest
+from collections import namedtuple
+from contextlib import contextmanager
+
+from django.db import connection
+from django.test import TestCase
+
+from ..models import Person
+
+
+@unittest.skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests')
+class ServerSideCursorsPostgres(TestCase):
+ cursor_fields = 'name, statement, is_holdable, is_binary, is_scrollable, creation_time'
+ PostgresCursor = namedtuple('PostgresCursor', cursor_fields)
+
+ @classmethod
+ def setUpTestData(cls):
+ Person.objects.create(first_name='a', last_name='a')
+ Person.objects.create(first_name='b', last_name='b')
+
+ def inspect_cursors(self):
+ with connection.cursor() as cursor:
+ cursor.execute('SELECT {fields} FROM pg_cursors;'.format(fields=self.cursor_fields))
+ cursors = cursor.fetchall()
+ return [self.PostgresCursor._make(cursor) for cursor in cursors]
+
+ @contextmanager
+ def override_db_setting(self, **kwargs):
+ for setting, value in kwargs.items():
+ original_value = connection.settings_dict.get(setting)
+ if setting in connection.settings_dict:
+ self.addCleanup(operator.setitem, connection.settings_dict, setting, original_value)
+ else:
+ self.addCleanup(operator.delitem, connection.settings_dict, setting)
+
+ connection.settings_dict[setting] = kwargs[setting]
+ yield
+
+ def test_server_side_cursor(self):
+ persons = Person.objects.iterator()
+ next(persons) # Open a server-side cursor
+ cursors = self.inspect_cursors()
+ self.assertEqual(len(cursors), 1)
+ self.assertIn('_django_curs_', cursors[0].name)
+ self.assertFalse(cursors[0].is_scrollable)
+ self.assertFalse(cursors[0].is_holdable)
+ self.assertFalse(cursors[0].is_binary)
+
+ def test_server_side_cursor_many_cursors(self):
+ persons = Person.objects.iterator()
+ persons2 = Person.objects.iterator()
+ next(persons) # Open a server-side cursor
+ next(persons2) # Open a second server-side cursor
+ cursors = self.inspect_cursors()
+ self.assertEqual(len(cursors), 2)
+ for cursor in cursors:
+ self.assertIn('_django_curs_', cursor.name)
+ self.assertFalse(cursor.is_scrollable)
+ self.assertFalse(cursor.is_holdable)
+ self.assertFalse(cursor.is_binary)
+
+ def test_closed_server_side_cursor(self):
+ persons = Person.objects.iterator()
+ next(persons) # Open a server-side cursor
+ del persons
+ cursors = self.inspect_cursors()
+ self.assertEqual(len(cursors), 0)
+
+ def test_server_side_cursors_setting(self):
+ with self.override_db_setting(DISABLE_SERVER_SIDE_CURSORS=False):
+ persons = Person.objects.iterator()
+ next(persons) # Open a server-side cursor
+ cursors = self.inspect_cursors()
+ self.assertEqual(len(cursors), 1)
+ del persons # Close server-side cursor
+
+ with self.override_db_setting(DISABLE_SERVER_SIDE_CURSORS=True):
+ persons = Person.objects.iterator()
+ next(persons) # Should not open a server-side cursor
+ cursors = self.inspect_cursors()
+ self.assertEqual(len(cursors), 0)
diff --git a/tests/backends/postgresql/tests.py b/tests/backends/postgresql/tests.py
new file mode 100644
index 0000000000..140fbbc444
--- /dev/null
+++ b/tests/backends/postgresql/tests.py
@@ -0,0 +1,147 @@
+import unittest
+import warnings
+from unittest import mock
+
+from django.db import DatabaseError, connection
+from django.test import TestCase
+
+
+@unittest.skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests')
+class Tests(TestCase):
+
+ def test_nodb_connection(self):
+ """
+ The _nodb_connection property fallbacks to the default connection
+ database when access to the 'postgres' database is not granted.
+ """
+ def mocked_connect(self):
+ if self.settings_dict['NAME'] is None:
+ raise DatabaseError()
+ return ''
+
+ nodb_conn = connection._nodb_connection
+ self.assertIsNone(nodb_conn.settings_dict['NAME'])
+
+ # Now assume the 'postgres' db isn't available
+ with warnings.catch_warnings(record=True) as w:
+ with mock.patch('django.db.backends.base.base.BaseDatabaseWrapper.connect',
+ side_effect=mocked_connect, autospec=True):
+ warnings.simplefilter('always', RuntimeWarning)
+ nodb_conn = connection._nodb_connection
+ self.assertIsNotNone(nodb_conn.settings_dict['NAME'])
+ self.assertEqual(nodb_conn.settings_dict['NAME'], connection.settings_dict['NAME'])
+ # Check a RuntimeWarning has been emitted
+ self.assertEqual(len(w), 1)
+ self.assertEqual(w[0].message.__class__, RuntimeWarning)
+
+ def test_connect_and_rollback(self):
+ """
+ PostgreSQL shouldn't roll back SET TIME ZONE, even if the first
+ transaction is rolled back (#17062).
+ """
+ new_connection = connection.copy()
+ try:
+ # Ensure the database default time zone is different than
+ # the time zone in new_connection.settings_dict. We can
+ # get the default time zone by reset & show.
+ cursor = new_connection.cursor()
+ cursor.execute("RESET TIMEZONE")
+ cursor.execute("SHOW TIMEZONE")
+ db_default_tz = cursor.fetchone()[0]
+ new_tz = 'Europe/Paris' if db_default_tz == 'UTC' else 'UTC'
+ new_connection.close()
+
+ # Invalidate timezone name cache, because the setting_changed
+ # handler cannot know about new_connection.
+ del new_connection.timezone_name
+
+ # Fetch a new connection with the new_tz as default
+ # time zone, run a query and rollback.
+ with self.settings(TIME_ZONE=new_tz):
+ new_connection.set_autocommit(False)
+ cursor = new_connection.cursor()
+ new_connection.rollback()
+
+ # Now let's see if the rollback rolled back the SET TIME ZONE.
+ cursor.execute("SHOW TIMEZONE")
+ tz = cursor.fetchone()[0]
+ self.assertEqual(new_tz, tz)
+
+ finally:
+ new_connection.close()
+
+ def test_connect_non_autocommit(self):
+ """
+ The connection wrapper shouldn't believe that autocommit is enabled
+ after setting the time zone when AUTOCOMMIT is False (#21452).
+ """
+ new_connection = connection.copy()
+ new_connection.settings_dict['AUTOCOMMIT'] = False
+
+ try:
+ # Open a database connection.
+ new_connection.cursor()
+ self.assertFalse(new_connection.get_autocommit())
+ finally:
+ new_connection.close()
+
+ def test_connect_isolation_level(self):
+ """
+ The transaction level can be configured with
+ DATABASES ['OPTIONS']['isolation_level'].
+ """
+ import psycopg2
+ from psycopg2.extensions import (
+ ISOLATION_LEVEL_READ_COMMITTED as read_committed,
+ ISOLATION_LEVEL_SERIALIZABLE as serializable,
+ )
+ # Since this is a django.test.TestCase, a transaction is in progress
+ # and the isolation level isn't reported as 0. This test assumes that
+ # PostgreSQL is configured with the default isolation level.
+
+ # Check the level on the psycopg2 connection, not the Django wrapper.
+ default_level = read_committed if psycopg2.__version__ < '2.7' else None
+ self.assertEqual(connection.connection.isolation_level, default_level)
+
+ new_connection = connection.copy()
+ new_connection.settings_dict['OPTIONS']['isolation_level'] = serializable
+ try:
+ # Start a transaction so the isolation level isn't reported as 0.
+ new_connection.set_autocommit(False)
+ # Check the level on the psycopg2 connection, not the Django wrapper.
+ self.assertEqual(new_connection.connection.isolation_level, serializable)
+ finally:
+ new_connection.close()
+
+ def _select(self, val):
+ with connection.cursor() as cursor:
+ cursor.execute('SELECT %s', (val,))
+ return cursor.fetchone()[0]
+
+ def test_select_ascii_array(self):
+ a = ['awef']
+ b = self._select(a)
+ self.assertEqual(a[0], b[0])
+
+ def test_select_unicode_array(self):
+ a = ['ᄲawef']
+ b = self._select(a)
+ self.assertEqual(a[0], b[0])
+
+ def test_lookup_cast(self):
+ from django.db.backends.postgresql.operations import DatabaseOperations
+ do = DatabaseOperations(connection=None)
+ lookups = (
+ 'iexact', 'contains', 'icontains', 'startswith', 'istartswith',
+ 'endswith', 'iendswith', 'regex', 'iregex',
+ )
+ for lookup in lookups:
+ with self.subTest(lookup=lookup):
+ self.assertIn('::text', do.lookup_cast(lookup))
+
+ def test_correct_extraction_psycopg2_version(self):
+ from django.db.backends.postgresql.base import psycopg2_version
+ with mock.patch('psycopg2.__version__', '4.2.1 (dt dec pq3 ext lo64)'):
+ self.assertEqual(psycopg2_version(), (4, 2, 1))
+ with mock.patch('psycopg2.__version__', '4.2b0.dev1 (dt dec pq3 ext lo64)'):
+ self.assertEqual(psycopg2_version(), (4, 2))
generated by cgit v1.3 (git 2.53.0) at 2026-08-16 10:14:24 +0000