Fix test creation tests.

django/db/backends/postgresql/base.py

@@ -1,7 +1,7 @@
 """
 PostgreSQL database backend for Django.
 
-Requires psycopg > 2: https://www.psycopg.org/psycopg3/
+Requires psycopg2 >= 2.8.4 or psycopg3
 """
 
 import asyncio
@@ -11,11 +11,10 @@
 
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
-from django.db import DatabaseError as WrappedDatabaseError, connections
+from django.db import DatabaseError as WrappedDatabaseError
+from django.db import connections
 from django.db.backends.base.base import BaseDatabaseWrapper
-from django.db.backends.utils import (
-    CursorDebugWrapper as BaseCursorDebugWrapper,
-)
+from django.db.backends.utils import CursorDebugWrapper as BaseCursorDebugWrapper
 from django.db.utils import Text
 from django.utils.asyncio import async_unsafe
 from django.utils.functional import cached_property
@@ -39,13 +38,20 @@
 
 
 def psycopg_version():
-    version = Database.__version__
+    version = Database.__version__.split(" ", 1)[0]
     return get_version_tuple(version)
 
 
 PSYCOPG_VERSION = psycopg_version()
 
-# Some of these import psycopg3, so import them after checking if it's installed.
+
+if PSYCOPG_VERSION < (2, 8, 4):
+    raise ImproperlyConfigured(
+        "psycopg2 version 2.8.4 or newer is required; you have %s"
+        % Database.__version__
+    )
+
+# Some of these import psycopg, so import them after checking if it's installed.
 from .client import DatabaseClient  # NOQA isort:skip
 from .creation import DatabaseCreation  # NOQA isort:skip
 from .features import DatabaseFeatures  # NOQA isort:skip
@@ -225,8 +231,8 @@ def get_new_connection(self, conn_params):
                 self.isolation_level = Database.IsolationLevel(isolevel)
             except ValueError:
                 raise ImproperlyConfigured(
-                    "bad isolation_level: %s. Choose one of the 'psycopg.IsolationLevel' values"
-                    % (options["isolation_level"],)
+                    "bad isolation_level: %s. Choose one of the "
+                    "'psycopg.IsolationLevel' values" % (options["isolation_level"],)
                 )
             connection.isolation_level = self.isolation_level
 
@@ -314,12 +320,7 @@ def chunked_cursor(self):
         # For now, it's here so that every use of "threading" is
         # also async-compatible.
         try:
-            if hasattr(asyncio, "current_task"):
-                # Python 3.7 and up
-                current_task = asyncio.current_task()
-            else:
-                # Python 3.6
-                current_task = asyncio.Task.current_task()
+            current_task = asyncio.current_task()
         except RuntimeError:
             current_task = None
         # Current task can be none even if the current_task call didn't error

django/db/backends/postgresql/creation.py

@@ -1,11 +1,14 @@
 import sys
 
-from psycopg import errors
-
 from django.core.exceptions import ImproperlyConfigured
 from django.db.backends.base.creation import BaseDatabaseCreation
 from django.db.backends.utils import strip_quotes
 
+try:
+    from psycopg import errors
+except ImportError:
+    from psycopg2 import errors
+
 
 class DatabaseCreation(BaseDatabaseCreation):
     def _quote_name(self, name):
@@ -46,12 +49,12 @@ def _execute_create_test_db(self, cursor, parameters, keepdb=False):
                 return
             super()._execute_create_test_db(cursor, parameters, keepdb)
         except Exception as e:
-            sqlstate = getattr(e.__cause__, "sqlstate", None)
-            if sqlstate != errors.DuplicateDatabase.sqlstate:
+            cause = e.__cause__
+            if cause and not isinstance(cause, errors.DuplicateDatabase):
                 # All errors except "database already exists" cancel tests.
                 self.log("Got an error creating the test database: %s" % e)
                 sys.exit(2)
-            elif not keepdb:
+            if not keepdb:
                 # If the database should be kept, ignore "database already
                 # exists".
                 raise

tests/backends/postgresql/test_creation.py

@@ -3,39 +3,17 @@
 from io import StringIO
 from unittest import mock
 
-from django.core.exceptions import ImproperlyConfigured
 from django.conf import settings
-from django.db import DatabaseError, connection, DEFAULT_DB_ALIAS
+from django.core.exceptions import ImproperlyConfigured
+from django.db import DEFAULT_DB_ALIAS, DatabaseError, connection
 from django.db.backends.base.creation import BaseDatabaseCreation
 from django.test import SimpleTestCase
 from django.utils.module_loading import import_string
 
-psycopg2 = None
-try:
-    import psycopg2  # NOQA
-except ImportError:
-    pass
-else:
-    import psycopg2.errorcodes
-
-
-psycopg = None
 try:
-    import psycopg
+    from psycopg import errors as psycopg_errors
 except ImportError:
-    pass
-else:
-    import psycopg.errors
-
-
-def set_pgcode(error, const_name):
-    """Hack up an exception to resemble one raised by psycopg."""
-    if psycopg2:
-        error.pgcode = getattr(psycopg2.errorcodes, const_name)
-    elif psycopg:
-        error.sqlstate = psycopg.errors.lookup(const_name).sqlstate
-    else:
-        raise ImportError("no psycopg module available")
+    from psycopg2 import errors as psycopg_errors
 
 
 @unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL tests")
@@ -93,13 +71,15 @@ def test_sql_table_creation_raises_with_collation(self):
             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"])
-        set_pgcode(error, "DUPLICATE_DATABASE")
+        error = psycopg_errors.DuplicateDatabase(
+            "database %s already exists" % parameters["dbname"]
+        )
         raise DatabaseError() from error
 
     def _execute_raise_permission_denied(self, cursor, parameters, keepdb=False):
-        error = DatabaseError("permission denied to create database")
-        set_pgcode(error, "INSUFFICIENT_PRIVILEGE")
+        error = psycopg_errors.InsufficientPrivilege(
+            "permission denied to create database"
+        )
         raise DatabaseError() from error
 
     def patch_test_db_creation(self, execute_create_test_db):