diff options
Diffstat (limited to 'tests/backends/postgresql')
| -rw-r--r-- | tests/backends/postgresql/test_creation.py | 57 | ||||
| -rw-r--r-- | tests/backends/postgresql/test_introspection.py | 14 | ||||
| -rw-r--r-- | tests/backends/postgresql/test_operations.py | 2 | ||||
| -rw-r--r-- | tests/backends/postgresql/test_server_side_cursors.py | 38 | ||||
| -rw-r--r-- | tests/backends/postgresql/tests.py | 115 |
5 files changed, 138 insertions, 88 deletions
diff --git a/tests/backends/postgresql/test_creation.py b/tests/backends/postgresql/test_creation.py index eb68c6e471..319029334d 100644 --- a/tests/backends/postgresql/test_creation.py +++ b/tests/backends/postgresql/test_creation.py @@ -18,12 +18,11 @@ else: from django.db.backends.postgresql.creation import DatabaseCreation -@unittest.skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests') +@unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL tests") class DatabaseCreationTests(SimpleTestCase): - @contextmanager def changed_test_settings(self, **kwargs): - settings = connection.settings_dict['TEST'] + settings = connection.settings_dict["TEST"] saved_values = {} for name in kwargs: if name in settings: @@ -47,66 +46,80 @@ class DatabaseCreationTests(SimpleTestCase): self.assertEqual(suffix, expected) def test_sql_table_creation_suffix_with_none_settings(self): - settings = {'CHARSET': None, 'TEMPLATE': None} + settings = {"CHARSET": None, "TEMPLATE": None} self.check_sql_table_creation_suffix(settings, "") def test_sql_table_creation_suffix_with_encoding(self): - settings = {'CHARSET': 'UTF8'} + 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'} + 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"''') + settings = {"CHARSET": "UTF8", "TEMPLATE": "template0"} + self.check_sql_table_creation_suffix( + settings, '''WITH ENCODING 'UTF8' TEMPLATE "template0"''' + ) def test_sql_table_creation_raises_with_collation(self): - settings = {'COLLATION': 'test'} + settings = {"COLLATION": "test"} msg = ( - 'PostgreSQL does not support collation setting at database ' - 'creation time.' + "PostgreSQL does not support collation setting at database " + "creation time." ) with self.assertRaisesMessage(ImproperlyConfigured, msg): self.check_sql_table_creation_suffix(settings, None) def _execute_raise_database_already_exists(self, cursor, parameters, keepdb=False): - error = DatabaseError('database %s already exists' % parameters['dbname']) + 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 = 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) + 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) + @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 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) + 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 mock.patch.object(DatabaseCreation, '_database_exists', return_value=False): + with mock.patch.object( + DatabaseCreation, "_database_exists", return_value=False + ): with self.assertRaises(SystemExit): - creation._create_test_db(verbosity=0, autoclobber=False, keepdb=False) + creation._create_test_db( + verbosity=0, autoclobber=False, keepdb=False + ) with self.assertRaises(SystemExit): - creation._create_test_db(verbosity=0, autoclobber=False, keepdb=True) + creation._create_test_db( + verbosity=0, autoclobber=False, keepdb=True + ) # Simulate test database creation raising "insufficient privileges". # An error shouldn't appear when keepdb is on and the database already # exists. with self.patch_test_db_creation(self._execute_raise_permission_denied): - with mock.patch.object(DatabaseCreation, '_database_exists', return_value=True): + with mock.patch.object( + DatabaseCreation, "_database_exists", return_value=True + ): creation._create_test_db(verbosity=0, autoclobber=False, keepdb=True) diff --git a/tests/backends/postgresql/test_introspection.py b/tests/backends/postgresql/test_introspection.py index 4dcadbd733..dc95d6ad23 100644 --- a/tests/backends/postgresql/test_introspection.py +++ b/tests/backends/postgresql/test_introspection.py @@ -6,18 +6,24 @@ from django.test import TestCase from ..models import Person -@unittest.skipUnless(connection.vendor == 'postgresql', "Test only for PostgreSQL") +@unittest.skipUnless(connection.vendor == "postgresql", "Test only for PostgreSQL") class DatabaseSequenceTests(TestCase): def test_get_sequences(self): with connection.cursor() as cursor: seqs = connection.introspection.get_sequences(cursor, Person._meta.db_table) self.assertEqual( seqs, - [{'table': Person._meta.db_table, 'column': 'id', 'name': 'backends_person_id_seq'}] + [ + { + "table": Person._meta.db_table, + "column": "id", + "name": "backends_person_id_seq", + } + ], ) - cursor.execute('ALTER SEQUENCE backends_person_id_seq RENAME TO pers_seq') + cursor.execute("ALTER SEQUENCE backends_person_id_seq RENAME TO pers_seq") seqs = connection.introspection.get_sequences(cursor, Person._meta.db_table) self.assertEqual( seqs, - [{'table': Person._meta.db_table, 'column': 'id', 'name': 'pers_seq'}] + [{"table": Person._meta.db_table, "column": "id", "name": "pers_seq"}], ) diff --git a/tests/backends/postgresql/test_operations.py b/tests/backends/postgresql/test_operations.py index 821bb29cee..c2f2417923 100644 --- a/tests/backends/postgresql/test_operations.py +++ b/tests/backends/postgresql/test_operations.py @@ -7,7 +7,7 @@ from django.test import SimpleTestCase from ..models import Person, Tag -@unittest.skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests.') +@unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL tests.") class PostgreSQLOperationsTests(SimpleTestCase): def test_sql_flush(self): self.assertEqual( diff --git a/tests/backends/postgresql/test_server_side_cursors.py b/tests/backends/postgresql/test_server_side_cursors.py index 0cc3423a9b..705e798c23 100644 --- a/tests/backends/postgresql/test_server_side_cursors.py +++ b/tests/backends/postgresql/test_server_side_cursors.py @@ -9,19 +9,23 @@ from django.test import TestCase from ..models import Person -@unittest.skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests') +@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) + 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') + 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)) + cursor.execute( + "SELECT {fields} FROM pg_cursors;".format(fields=self.cursor_fields) + ) cursors = cursor.fetchall() return [self.PostgresCursor._make(cursor) for cursor in cursors] @@ -30,7 +34,9 @@ class ServerSideCursorsPostgres(TestCase): for setting in kwargs: original_value = connection.settings_dict.get(setting) if setting in connection.settings_dict: - self.addCleanup(operator.setitem, connection.settings_dict, setting, original_value) + self.addCleanup( + operator.setitem, connection.settings_dict, setting, original_value + ) else: self.addCleanup(operator.delitem, connection.settings_dict, setting) @@ -42,7 +48,7 @@ class ServerSideCursorsPostgres(TestCase): cursors = self.inspect_cursors() self.assertEqual(len(cursors), num_expected) for cursor in cursors: - self.assertIn('_django_curs_', cursor.name) + self.assertIn("_django_curs_", cursor.name) self.assertFalse(cursor.is_scrollable) self.assertFalse(cursor.is_holdable) self.assertFalse(cursor.is_binary) @@ -54,17 +60,23 @@ class ServerSideCursorsPostgres(TestCase): self.assertUsesCursor(Person.objects.iterator()) def test_values(self): - self.assertUsesCursor(Person.objects.values('first_name').iterator()) + self.assertUsesCursor(Person.objects.values("first_name").iterator()) def test_values_list(self): - self.assertUsesCursor(Person.objects.values_list('first_name').iterator()) + self.assertUsesCursor(Person.objects.values_list("first_name").iterator()) def test_values_list_flat(self): - self.assertUsesCursor(Person.objects.values_list('first_name', flat=True).iterator()) + self.assertUsesCursor( + Person.objects.values_list("first_name", flat=True).iterator() + ) def test_values_list_fields_not_equal_to_names(self): - expr = models.Count('id') - self.assertUsesCursor(Person.objects.annotate(id__count=expr).values_list(expr, 'id__count').iterator()) + expr = models.Count("id") + self.assertUsesCursor( + Person.objects.annotate(id__count=expr) + .values_list(expr, "id__count") + .iterator() + ) def test_server_side_cursor_many_cursors(self): persons = Person.objects.iterator() diff --git a/tests/backends/postgresql/tests.py b/tests/backends/postgresql/tests.py index 1905147f6f..af08f6f286 100644 --- a/tests/backends/postgresql/tests.py +++ b/tests/backends/postgresql/tests.py @@ -9,9 +9,9 @@ from django.db.backends.base.base import BaseDatabaseWrapper from django.test import TestCase, override_settings -@unittest.skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests') +@unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL tests") class Tests(TestCase): - databases = {'default', 'other'} + databases = {"default", "other"} def test_nodb_cursor(self): """ @@ -21,14 +21,14 @@ class Tests(TestCase): orig_connect = BaseDatabaseWrapper.connect def mocked_connect(self): - if self.settings_dict['NAME'] is None: + if self.settings_dict["NAME"] is None: raise DatabaseError() return orig_connect(self) with connection._nodb_cursor() as cursor: self.assertIs(cursor.closed, False) self.assertIsNotNone(cursor.db.connection) - self.assertIsNone(cursor.db.settings_dict['NAME']) + self.assertIsNone(cursor.db.settings_dict["NAME"]) self.assertIs(cursor.closed, True) self.assertIsNone(cursor.db.connection) @@ -41,24 +41,29 @@ class Tests(TestCase): "database and will use the first PostgreSQL database instead." ) with self.assertWarnsMessage(RuntimeWarning, msg): - with mock.patch('django.db.backends.base.base.BaseDatabaseWrapper.connect', - side_effect=mocked_connect, autospec=True): + with mock.patch( + "django.db.backends.base.base.BaseDatabaseWrapper.connect", + side_effect=mocked_connect, + autospec=True, + ): with mock.patch.object( connection, - 'settings_dict', - {**connection.settings_dict, 'NAME': 'postgres'}, + "settings_dict", + {**connection.settings_dict, "NAME": "postgres"}, ): 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']) + self.assertIsNotNone(cursor.db.settings_dict["NAME"]) + self.assertEqual( + cursor.db.settings_dict["NAME"], connections["other"].settings_dict["NAME"] + ) # Cursor is yielded only for the first PostgreSQL database. with self.assertWarnsMessage(RuntimeWarning, msg): with mock.patch( - 'django.db.backends.base.base.BaseDatabaseWrapper.connect', + "django.db.backends.base.base.BaseDatabaseWrapper.connect", side_effect=mocked_connect, autospec=True, ): @@ -71,13 +76,14 @@ class Tests(TestCase): _nodb_cursor() re-raises authentication failure to the 'postgres' db when other connection to the PostgreSQL database isn't available. """ + def mocked_connect(self): raise DatabaseError() def mocked_all(self): test_connection = copy.copy(connections[DEFAULT_DB_ALIAS]) test_connection.settings_dict = copy.deepcopy(connection.settings_dict) - test_connection.settings_dict['NAME'] = 'postgres' + test_connection.settings_dict["NAME"] = "postgres" return [test_connection] msg = ( @@ -89,12 +95,12 @@ class Tests(TestCase): ) with self.assertWarnsMessage(RuntimeWarning, msg): mocker_connections_all = mock.patch( - 'django.utils.connection.BaseConnectionHandler.all', + "django.utils.connection.BaseConnectionHandler.all", side_effect=mocked_all, autospec=True, ) mocker_connect = mock.patch( - 'django.db.backends.base.base.BaseDatabaseWrapper.connect', + "django.db.backends.base.base.BaseDatabaseWrapper.connect", side_effect=mocked_connect, autospec=True, ) @@ -104,27 +110,29 @@ class Tests(TestCase): pass def test_nodb_cursor_reraise_exceptions(self): - with self.assertRaisesMessage(DatabaseError, 'exception'): + with self.assertRaisesMessage(DatabaseError, "exception"): with connection._nodb_cursor(): - raise DatabaseError('exception') + raise DatabaseError("exception") def test_database_name_too_long(self): from django.db.backends.postgresql.base import DatabaseWrapper + settings = connection.settings_dict.copy() max_name_length = connection.ops.max_name_length() - settings['NAME'] = 'a' + (max_name_length * 'a') + settings["NAME"] = "a" + (max_name_length * "a") msg = ( "The database name '%s' (%d characters) is longer than " "PostgreSQL's limit of %s characters. Supply a shorter NAME in " "settings.DATABASES." - ) % (settings['NAME'], max_name_length + 1, max_name_length) + ) % (settings["NAME"], max_name_length + 1, max_name_length) with self.assertRaisesMessage(ImproperlyConfigured, msg): DatabaseWrapper(settings).get_connection_params() def test_database_name_empty(self): from django.db.backends.postgresql.base import DatabaseWrapper + settings = connection.settings_dict.copy() - settings['NAME'] = '' + settings["NAME"] = "" msg = ( "settings.DATABASES is improperly configured. Please supply the " "NAME or OPTIONS['service'] value." @@ -134,22 +142,24 @@ class Tests(TestCase): def test_service_name(self): from django.db.backends.postgresql.base import DatabaseWrapper + settings = connection.settings_dict.copy() - settings['OPTIONS'] = {'service': 'my_service'} - settings['NAME'] = '' + settings["OPTIONS"] = {"service": "my_service"} + settings["NAME"] = "" params = DatabaseWrapper(settings).get_connection_params() - self.assertEqual(params['service'], 'my_service') - self.assertNotIn('database', params) + self.assertEqual(params["service"], "my_service") + self.assertNotIn("database", params) def test_service_name_default_db(self): # None is used to connect to the default 'postgres' db. from django.db.backends.postgresql.base import DatabaseWrapper + settings = connection.settings_dict.copy() - settings['NAME'] = None - settings['OPTIONS'] = {'service': 'django_test'} + settings["NAME"] = None + settings["OPTIONS"] = {"service": "django_test"} params = DatabaseWrapper(settings).get_connection_params() - self.assertEqual(params['database'], 'postgres') - self.assertNotIn('service', params) + self.assertEqual(params["database"], "postgres") + self.assertNotIn("service", params) def test_connect_and_rollback(self): """ @@ -165,7 +175,7 @@ class Tests(TestCase): 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_tz = "Europe/Paris" if db_default_tz == "UTC" else "UTC" new_connection.close() # Invalidate timezone name cache, because the setting_changed @@ -193,7 +203,7 @@ class Tests(TestCase): after setting the time zone when AUTOCOMMIT is False (#21452). """ new_connection = connection.copy() - new_connection.settings_dict['AUTOCOMMIT'] = False + new_connection.settings_dict["AUTOCOMMIT"] = False try: # Open a database connection. @@ -207,9 +217,7 @@ class Tests(TestCase): The transaction level can be configured with DATABASES ['OPTIONS']['isolation_level']. """ - from psycopg2.extensions import ( - ISOLATION_LEVEL_SERIALIZABLE as serializable, - ) + from psycopg2.extensions import 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 @@ -218,7 +226,7 @@ class Tests(TestCase): self.assertIsNone(connection.connection.isolation_level) new_connection = connection.copy() - new_connection.settings_dict['OPTIONS']['isolation_level'] = serializable + 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) @@ -230,7 +238,7 @@ class Tests(TestCase): def test_connect_no_is_usable_checks(self): new_connection = connection.copy() try: - with mock.patch.object(new_connection, 'is_usable') as is_usable: + with mock.patch.object(new_connection, "is_usable") as is_usable: new_connection.connect() is_usable.assert_not_called() finally: @@ -238,49 +246,60 @@ class Tests(TestCase): def _select(self, val): with connection.cursor() as cursor: - cursor.execute('SELECT %s', (val,)) + cursor.execute("SELECT %s", (val,)) return cursor.fetchone()[0] def test_select_ascii_array(self): - a = ['awef'] + a = ["awef"] b = self._select(a) self.assertEqual(a[0], b[0]) def test_select_unicode_array(self): - a = ['ᄲawef'] + 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', + "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)) + self.assertIn("::text", do.lookup_cast(lookup)) for lookup in lookups: - for field_type in ('CICharField', 'CIEmailField', 'CITextField'): + for field_type in ("CICharField", "CIEmailField", "CITextField"): with self.subTest(lookup=lookup, field_type=field_type): - self.assertIn('::citext', do.lookup_cast(lookup, internal_type=field_type)) + self.assertIn( + "::citext", do.lookup_cast(lookup, internal_type=field_type) + ) 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)'): + + 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)'): + with mock.patch("psycopg2.__version__", "4.2b0.dev1 (dt dec pq3 ext lo64)"): self.assertEqual(psycopg2_version(), (4, 2)) @override_settings(DEBUG=True) def test_copy_cursors(self): out = StringIO() - copy_expert_sql = 'COPY django_session TO STDOUT (FORMAT CSV, HEADER)' + copy_expert_sql = "COPY django_session TO STDOUT (FORMAT CSV, HEADER)" with connection.cursor() as cursor: cursor.copy_expert(copy_expert_sql, out) - cursor.copy_to(out, 'django_session') + cursor.copy_to(out, "django_session") self.assertEqual( - [q['sql'] for q in connection.queries], - [copy_expert_sql, 'COPY django_session TO STDOUT'], + [q["sql"] for q in connection.queries], + [copy_expert_sql, "COPY django_session TO STDOUT"], ) |