/project/testbed

./docs/man/django-admin.1:273:.SS \fBdbshell\fP
./docs/man/django-admin.1:276:.B django\-admin dbshell
./docs/releases/2.2.txt:128:* On Oracle, :djadmin:`dbshell` is wrapped with ``rlwrap``, if available.
./django/db/backends/postgresql/client.py:21:        sslrootcert = conn_params.get('sslrootcert', '')
./django/db/backends/postgresql/client.py:22:        sslcert = conn_params.get('sslcert', '')
./django/db/backends/postgresql/client.py:23:        sslkey = conn_params.get('sslkey', '')
./django/db/backends/postgresql/client.py:39:        if sslrootcert:
./django/db/backends/postgresql/client.py:40:            subprocess_env['PGSSLROOTCERT'] = str(sslrootcert)
./django/db/backends/postgresql/client.py:41:        if sslcert:
./django/db/backends/postgresql/client.py:42:            subprocess_env['PGSSLCERT'] = str(sslcert)
./django/db/backends/postgresql/client.py:43:        if sslkey:
./django/db/backends/postgresql/client.py:44:            subprocess_env['PGSSLKEY'] = str(sslkey)
./docs/releases/1.8.txt:436:* The :djadmin:`dbshell` command now supports MySQL's optional SSL certificate
./docs/releases/1.9.txt:440:* When using the PostgreSQL backend, the :djadmin:`dbshell` command can connect
./docs/releases/1.9.7.txt:27:* Fixed ``dbshell`` crash on PostgreSQL with an empty database name
./docs/releases/2.2.1.txt:13:  database user password when using :djadmin:`dbshell` on Oracle
./docs/releases/2.0.txt:225:* On MySQL, :djadmin:`dbshell` now supports client-side TLS certificates.
./docs/ref/django-admin.txt:202:``dbshell``
./docs/ref/django-admin.txt:205:.. django-admin:: dbshell
./tests/dbshell/test_postgresql.py:95:        # dbshell restores the original handler.
./tests/dbshell/test_mysql.py:63:             '--ssl-cert=sslcert', '--ssl-key=sslkey', 'somedbname'],
./tests/dbshell/test_mysql.py:73:                        'cert': 'sslcert',
./tests/dbshell/test_mysql.py:74:                        'key': 'sslkey',
./tests/dbshell/test_oracle.py:10:    def _run_dbshell(self, rlwrap=False):
./tests/dbshell/test_oracle.py:25:            self._run_dbshell(rlwrap=False),
./tests/dbshell/test_oracle.py:31:            self._run_dbshell(rlwrap=True),

import os
import signal
import subprocess

from django.db.backends.base.client import BaseDatabaseClient


class DatabaseClient(BaseDatabaseClient):
    executable_name = 'psql'

    @classmethod
    def runshell_db(cls, conn_params):
        args = [cls.executable_name]

        host = conn_params.get('host', '')
        port = conn_params.get('port', '')
        dbname = conn_params.get('database', '')
        user = conn_params.get('user', '')
        passwd = conn_params.get('password', '')
        sslmode = conn_params.get('sslmode', '')
        sslrootcert = conn_params.get('sslrootcert', '')
        sslcert = conn_params.get('sslcert', '')
        sslkey = conn_params.get('sslkey', '')

        if user:
            args += ['-U', user]
        if host:
            args += ['-h', host]
        if port:
            args += ['-p', str(port)]
        args += [dbname]

        sigint_handler = signal.getsignal(signal.SIGINT)
        subprocess_env = os.environ.copy()
        if passwd:
            subprocess_env['PGPASSWORD'] = str(passwd)
        if sslmode:
            subprocess_env['PGSSLMODE'] = str(sslmode)
        if sslrootcert:
            subprocess_env['PGSSLROOTCERT'] = str(sslrootcert)
        if sslcert:
            subprocess_env['PGSSLCERT'] = str(sslcert)
        if sslkey:
            subprocess_env['PGSSLKEY'] = str(sslkey)
        try:
            # Allow SIGINT to pass to psql to abort queries.
            signal.signal(signal.SIGINT, signal.SIG_IGN)
            subprocess.run(args, check=True, env=subprocess_env)
        finally:
            # Restore the original SIGINT handler.
            signal.signal(signal.SIGINT, sigint_handler)

    def runshell(self):
        DatabaseClient.runshell_db(self.connection.get_connection_params())

---TESTS---
import os
import signal
import subprocess
from unittest import mock

from django.db.backends.postgresql.client import DatabaseClient
from django.test import SimpleTestCase


class PostgreSqlDbshellCommandTestCase(SimpleTestCase):

    def _run_it(self, dbinfo):
        """
        That function invokes the runshell command, while mocking
        subprocess.run(). It returns a 2-tuple with:
        - The command line list
        - The the value of the PGPASSWORD environment variable, or None.
        """
        def _mock_subprocess_run(*args, env=os.environ, **kwargs):
            self.subprocess_args = list(*args)
            self.pgpassword = env.get('PGPASSWORD')
            return subprocess.CompletedProcess(self.subprocess_args, 0)
        with mock.patch('subprocess.run', new=_mock_subprocess_run):
            DatabaseClient.runshell_db(dbinfo)
        return self.subprocess_args, self.pgpassword

    def test_basic(self):
        self.assertEqual(
            self._run_it({
                'database': 'dbname',
                'user': 'someuser',
                'password': 'somepassword',
                'host': 'somehost',
                'port': '444',
            }), (
                ['psql', '-U', 'someuser', '-h', 'somehost', '-p', '444', 'dbname'],
                'somepassword',
            )
        )

    def test_nopass(self):
        self.assertEqual(
            self._run_it({
                'database': 'dbname',
                'user': 'someuser',
                'host': 'somehost',
                'port': '444',
            }), (
                ['psql', '-U', 'someuser', '-h', 'somehost', '-p', '444', 'dbname'],
                None,
            )
        )

    def test_column(self):
        self.assertEqual(
            self._run_it({
                'database': 'dbname',
                'user': 'some:user',
                'password': 'some:password',
                'host': '::1',
                'port': '444',
            }), (
                ['psql', '-U', 'some:user', '-h', '::1', '-p', '444', 'dbname'],
                'some:password',
            )
        )

    def test_accent(self):
        username = 'rôle'
        password = 'sésame'
        self.assertEqual(
            self._run_it({
                'database': 'dbname',
                'user': username,
                'password': password,
                'host': 'somehost',
                'port': '444',
            }), (
                ['psql', '-U', username, '-h', 'somehost', '-p', '444', 'dbname'],
                password,
            )
        )

    def test_sigint_handler(self):
        """SIGINT is ignored in Python and passed to psql to abort quries."""
        def _mock_subprocess_run(*args, **kwargs):
            handler = signal.getsignal(signal.SIGINT)
            self.assertEqual(handler, signal.SIG_IGN)

        sigint_handler = signal.getsignal(signal.SIGINT)
        # The default handler isn't SIG_IGN.
        self.assertNotEqual(sigint_handler, signal.SIG_IGN)
        with mock.patch('subprocess.run', new=_mock_subprocess_run):
            DatabaseClient.runshell_db({})
        # dbshell restores the original handler.
        self.assertEqual(sigint_handler, signal.getsignal(signal.SIGINT))

"""
PostgreSQL database backend for Django.

Requires psycopg 2: http://initd.org/projects/psycopg2
"""

import threading
import warnings

from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.db import connections
from django.db.backends.base.base import BaseDatabaseWrapper
from django.db.utils import DatabaseError as WrappedDatabaseError
from django.utils.functional import cached_property
from django.utils.safestring import SafeString
from django.utils.version import get_version_tuple

try:
    import psycopg2 as Database
    import psycopg2.extensions
    import psycopg2.extras
except ImportError as e:
    raise ImproperlyConfigured("Error loading psycopg2 module: %s" % e)


def psycopg2_version():
    version = psycopg2.__version__.split(' ', 1)[0]
    return get_version_tuple(version)


PSYCOPG2_VERSION = psycopg2_version()

if PSYCOPG2_VERSION < (2, 5, 4):
    raise ImproperlyConfigured("psycopg2_version 2.5.4 or newer is required; you have %s" % psycopg2.__version__)


# Some of these import psycopg2, 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
from .introspection import DatabaseIntrospection            # NOQA isort:skip
from .operations import DatabaseOperations                  # NOQA isort:skip
from .schema import DatabaseSchemaEditor                    # NOQA isort:skip
from .utils import utc_tzinfo_factory                       # NOQA isort:skip

psycopg2.extensions.register_adapter(SafeString, psycopg2.extensions.QuotedString)
psycopg2.extras.register_uuid()

# Register support for inet[] manually so we don't have to handle the Inet()
# object on load all the time.
INETARRAY_OID = 1041
INETARRAY = psycopg2.extensions.new_array_type(
    (INETARRAY_OID,),
    'INETARRAY',
    psycopg2.extensions.UNICODE,
)
psycopg2.extensions.register_type(INETARRAY)


class DatabaseWrapper(BaseDatabaseWrapper):
    vendor = 'postgresql'
    display_name = 'PostgreSQL'
    # This dictionary maps Field objects to their associated PostgreSQL column
    # types, as strings. Column-type strings can contain format strings; they'll
    # be interpolated against the values of Field.__dict__ before being output.
    # If a column type is set to None, it won't be included in the output.
    data_types = {
        'AutoField': 'serial',
        'BigAutoField': 'bigserial',
        'BinaryField': 'bytea',
        'BooleanField': 'boolean',
        'CharField': 'varchar(%(max_length)s)',
        'DateField': 'date',
        'DateTimeField': 'timestamp with time zone',
        'DecimalField': 'numeric(%(max_digits)s, %(decimal_places)s)',
        'DurationField': 'interval',
        'FileField': 'varchar(%(max_length)s)',
        'FilePathField': 'varchar(%(max_length)s)',
        'FloatField': 'double precision',
        'IntegerField': 'integer',
        'BigIntegerField': 'bigint',
        'IPAddressField': 'inet',
        'GenericIPAddressField': 'inet',
        'NullBooleanField': 'boolean',
        'OneToOneField': 'integer',
        'PositiveIntegerField': 'integer',
        'PositiveSmallIntegerField': 'smallint',
        'SlugField': 'varchar(%(max_length)s)',
        'SmallIntegerField': 'smallint',
        'TextField': 'text',
        'TimeField': 'time',
        'UUIDField': 'uuid',
    }
    data_type_check_constraints = {
        'PositiveIntegerField': '"%(column)s" >= 0',
        'PositiveSmallIntegerField': '"%(column)s" >= 0',
    }
    operators = {
        'exact': '= %s',
        'iexact': '= UPPER(%s)',
        'contains': 'LIKE %s',
        'icontains': 'LIKE UPPER(%s)',
        'regex': '~ %s',
        'iregex': '~* %s',
        'gt': '> %s',
        'gte': '>= %s',
        'lt': '< %s',
        'lte': '<= %s',
        'startswith': 'LIKE %s',
        'endswith': 'LIKE %s',
        'istartswith': 'LIKE UPPER(%s)',
        'iendswith': 'LIKE UPPER(%s)',
    }

    # The patterns below are used to generate SQL pattern lookup clauses when
    # the right-hand side of the lookup isn't a raw string (it might be an expression
    # or the result of a bilateral transformation).
    # In those cases, special characters for LIKE operators (e.g. \, *, _) should be
    # escaped on database side.
    #
    # Note: we use str.format() here for readability as '%' is used as a wildcard for
    # the LIKE operator.
    pattern_esc = r"REPLACE(REPLACE(REPLACE({}, E'\\', E'\\\\'), E'%%', E'\\%%'), E'_', E'\\_')"
    pattern_ops = {
        'contains': "LIKE '%%' || {} || '%%'",
        'icontains': "LIKE '%%' || UPPER({}) || '%%'",
        'startswith': "LIKE {} || '%%'",
        'istartswith': "LIKE UPPER({}) || '%%'",
        'endswith': "LIKE '%%' || {}",
        'iendswith': "LIKE '%%' || UPPER({})",
    }

    Database = Database
    SchemaEditorClass = DatabaseSchemaEditor
    # Classes instantiated in __init__().
    client_class = DatabaseClient
    creation_class = DatabaseCreation
    features_class = DatabaseFeatures
    introspection_class = DatabaseIntrospection
    ops_class = DatabaseOperations
    # PostgreSQL backend-specific attributes.
    _named_cursor_idx = 0

    def get_connection_params(self):
        settings_dict = self.settings_dict
        # None may be used to connect to the default 'postgres' db
        if settings_dict['NAME'] == '':
            raise ImproperlyConfigured(
                "settings.DATABASES is improperly configured. "
                "Please supply the NAME value.")
        if len(settings_dict['NAME'] or '') > self.ops.max_name_length():
            raise ImproperlyConfigured(
                "The database name '%s' (%d characters) is longer than "
                "PostgreSQL's limit of %d characters. Supply a shorter NAME "
                "in settings.DATABASES." % (
                    settings_dict['NAME'],
                    len(settings_dict['NAME']),
                    self.ops.max_name_length(),
                )
            )
        conn_params = {
            'database': settings_dict['NAME'] or 'postgres',
            **settings_dict['OPTIONS'],
        }
        conn_params.pop('isolation_level', None)
        if settings_dict['USER']:
            conn_params['user'] = settings_dict['USER']
        if settings_dict['PASSWORD']:
            conn_params['password'] = settings_dict['PASSWORD']
        if settings_dict['HOST']:
            conn_params['host'] = settings_dict['HOST']
        if settings_dict['PORT']:
            conn_params['port'] = settings_dict['PORT']
        return conn_params

    def get_new_connection(self, conn_params):
        connection = Database.connect(**conn_params)

        # self.isolation_level must be set:
        # - after connecting to the database in order to obtain the database's
        #   default when no value is explicitly specified in options.
        # - before calling _set_autocommit() because if autocommit is on, that
        #   will set connection.isolation_level to ISOLATION_LEVEL_AUTOCOMMIT.
        options = self.settings_dict['OPTIONS']
        try:
            self.isolation_level = options['isolation_level']
        except KeyError:
            self.isolation_level = connection.isolation_level
        else:
            # Set the isolation level to the value from OPTIONS.
            if self.isolation_level != connection.isolation_level:
                connection.set_session(isolation_level=self.isolation_level)

        return connection

    def ensure_timezone(self):
        if self.connection is None:
            return False
        conn_timezone_name = self.connection.get_parameter_status('TimeZone')
        timezone_name = self.timezone_name
        if timezone_name and conn_timezone_name != timezone_name:
            with self.connection.cursor() as cursor:
                cursor.execute(self.ops.set_time_zone_sql(), [timezone_name])
            return True
        return False

    def init_connection_state(self):
        self.connection.set_client_encoding('UTF8')

        timezone_changed = self.ensure_timezone()
        if timezone_changed:
            # Commit after setting the time zone (see #17062)
            if not self.get_autocommit():
                self.connection.commit()

    def create_cursor(self, name=None):
        if name:
            # In autocommit mode, the cursor will be used outside of a
            # transaction, hence use a holdable cursor.

---BASE CLIENT---
class BaseDatabaseClient:
    """Encapsulate backend-specific methods for opening a client shell."""
    # This should be a string representing the name of the executable
    # (e.g., "psql"). Subclasses must override this.
    executable_name = None

    def __init__(self, connection):
        # connection is an instance of BaseDatabaseWrapper.
        self.connection = connection

    def runshell(self):
        raise NotImplementedError('subclasses of BaseDatabaseClient must provide a runshell() method')

/opt/miniconda3/bin/python: No module named pytest

AUTHORS
CONTRIBUTING.rst
Django.egg-info
Gruntfile.js
INSTALL
LICENSE
LICENSE.python
MANIFEST.in
README.rst
django
docs
extras
js_tests
package.json
scripts
setup.cfg
setup.py
tests
tox.ini
python: can't open file '/project/testbed/manage.py': [Errno 2] No such file or directory

django/db/backends/mysql/base.py:189:    def get_connection_params(self):
django/db/backends/oracle/base.py:218:    def get_connection_params(self):
django/db/backends/base/base.py:162:    def get_connection_params(self):
django/db/backends/base/base.py:164:        raise NotImplementedError('subclasses of BaseDatabaseWrapper may require a get_connection_params() method')
django/db/backends/base/base.py:194:        conn_params = self.get_connection_params()
django/db/backends/sqlite3/base.py:166:    def get_connection_params(self):
django/db/backends/postgresql/client.py:12:    def runshell_db(cls, conn_params):
django/db/backends/postgresql/client.py:21:        sslrootcert = conn_params.get('sslrootcert', '')
django/db/backends/postgresql/client.py:22:        sslcert = conn_params.get('sslcert', '')
django/db/backends/postgresql/client.py:23:        sslkey = conn_params.get('sslkey', '')
django/db/backends/postgresql/client.py:39:        if sslrootcert:
django/db/backends/postgresql/client.py:40:            subprocess_env['PGSSLROOTCERT'] = str(sslrootcert)
django/db/backends/postgresql/client.py:41:        if sslcert:
django/db/backends/postgresql/client.py:42:            subprocess_env['PGSSLCERT'] = str(sslcert)
django/db/backends/postgresql/client.py:43:        if sslkey:
django/db/backends/postgresql/client.py:44:            subprocess_env['PGSSLKEY'] = str(sslkey)
django/db/backends/postgresql/client.py:54:        DatabaseClient.runshell_db(self.connection.get_connection_params())
django/db/backends/postgresql/base.py:145:    def get_connection_params(self):
tests/backends/postgresql/tests.py:56:            DatabaseWrapper(settings).get_connection_params()
tests/dbshell/test_postgresql.py:24:            DatabaseClient.runshell_db(dbinfo)
tests/dbshell/test_postgresql.py:94:            DatabaseClient.runshell_db({})
tests/dbshell/test_mysql.py:63:             '--ssl-cert=sslcert', '--ssl-key=sslkey', 'somedbname'],
tests/dbshell/test_mysql.py:73:                        'cert': 'sslcert',
tests/dbshell/test_mysql.py:74:                        'key': 'sslkey',

---RUNTESTS---
sed: can't read runtests.py: No such file or directory

django/conf/project_template/manage.py-tpl
tests/project_template/test_settings.py
tests/auth_tests/settings.py
django/conf/global_settings.py
tests/dbshell/test_postgresql.py
tests/admin_scripts/configured_dynamic_settings_manage.py
tests/postgres_tests/integration_settings.py
tests/i18n/sampleproject/sampleproject/settings.py
tests/flatpages_tests/settings.py
tests/i18n/sampleproject/manage.py
tests/admin_scripts/custom_templates/project_template/manage.py-tpl
tests/admin_scripts/configured_settings_manage.py
tests/admin_scripts/custom_templates/project_template/project_name/settings.py
tests/runtests.py
tests/requests/test_data_upload_settings.py
tests/staticfiles_tests/settings.py
django/core/management/commands/diffsettings.py

---MYSQL TEST---
from django.db.backends.mysql.client import DatabaseClient
from django.test import SimpleTestCase


class MySqlDbshellCommandTestCase(SimpleTestCase):

    def test_fails_with_keyerror_on_incomplete_config(self):
        with self.assertRaises(KeyError):
            self.get_command_line_arguments({})

    def test_basic_params_specified_in_settings(self):
        self.assertEqual(
            ['mysql', '--user=someuser', '--password=somepassword',
             '--host=somehost', '--port=444', 'somedbname'],
            self.get_command_line_arguments({
                'NAME': 'somedbname',
                'USER': 'someuser',
                'PASSWORD': 'somepassword',
                'HOST': 'somehost',
                'PORT': 444,
                'OPTIONS': {},
            }))

    def test_options_override_settings_proper_values(self):
        settings_port = 444
        options_port = 555
        self.assertNotEqual(settings_port, options_port, 'test pre-req')
        self.assertEqual(
            ['mysql', '--user=optionuser', '--password=optionpassword',
             '--host=optionhost', '--port={}'.format(options_port), 'optiondbname'],
            self.get_command_line_arguments({
                'NAME': 'settingdbname',
                'USER': 'settinguser',
                'PASSWORD': 'settingpassword',
                'HOST': 'settinghost',
                'PORT': settings_port,
                'OPTIONS': {
                    'db': 'optiondbname',
                    'user': 'optionuser',
                    'passwd': 'optionpassword',
                    'host': 'optionhost',
                    'port': options_port,
                },
            }))

    def test_can_connect_using_sockets(self):
        self.assertEqual(
            ['mysql', '--user=someuser', '--password=somepassword',
             '--socket=/path/to/mysql.socket.file', 'somedbname'],
            self.get_command_line_arguments({
                'NAME': 'somedbname',
                'USER': 'someuser',
                'PASSWORD': 'somepassword',
                'HOST': '/path/to/mysql.socket.file',
                'PORT': None,
                'OPTIONS': {},
            }))

    def test_ssl_certificate_is_added(self):
        self.assertEqual(
            ['mysql', '--user=someuser', '--password=somepassword',
             '--host=somehost', '--port=444', '--ssl-ca=sslca',
             '--ssl-cert=sslcert', '--ssl-key=sslkey', 'somedbname'],
            self.get_command_line_arguments({
                'NAME': 'somedbname',
                'USER': 'someuser',
                'PASSWORD': 'somepassword',
                'HOST': 'somehost',
                'PORT': 444,
                'OPTIONS': {
                    'ssl': {
                        'ca': 'sslca',
                        'cert': 'sslcert',
                        'key': 'sslkey',
                    },
                },
            }))

    def get_command_line_arguments(self, connection_settings):
        return DatabaseClient.settings_to_cmd_args(connection_settings)

---BACKEND TESTS---
import unittest
from unittest import mock

from django.core.exceptions import ImproperlyConfigured
from django.db import DatabaseError, connection, connections
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
        msg = (
            "Normally Django will use a connection to the 'postgres' database "
            "to avoid running initialization queries against the production "
            "database when it's not needed (for example, when running tests). "
            "Django was unable to create a connection to the 'postgres' "
            "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.object(
                    connection,
                    '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'])

    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')
        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)
        with self.assertRaisesMessage(ImproperlyConfigured, msg):
            DatabaseWrapper(settings).get_connection_params()

    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.
            with new_connection.cursor() as 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)
                new_connection.rollback()

                # Now let's see if the rollback rolled back the SET TIME ZONE.
                with new_connection.cursor() as cursor:
                    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

['psql', '-U', 'user', '-h', 'host', '-p', '5432', 'dbname']
verify-ca
ca.crt
client.crt
client.key

     1	import os
     2	import signal
     3	import subprocess
     4	
     5	from django.db.backends.base.client import BaseDatabaseClient
     6	
     7	
     8	class DatabaseClient(BaseDatabaseClient):
     9	    executable_name = 'psql'
    10	
    11	    @classmethod
    12	    def runshell_db(cls, conn_params):
    13	        args = [cls.executable_name]
    14	
    15	        host = conn_params.get('host', '')
    16	        port = conn_params.get('port', '')
    17	        dbname = conn_params.get('database', '')
    18	        user = conn_params.get('user', '')
    19	        passwd = conn_params.get('password', '')
    20	        sslmode = conn_params.get('sslmode', '')
    21	        sslrootcert = conn_params.get('sslrootcert', '')
    22	        sslcert = conn_params.get('sslcert', '')
    23	        sslkey = conn_params.get('sslkey', '')
    24	
    25	        if user:
    26	            args += ['-U', user]
    27	        if host:
    28	            args += ['-h', host]
    29	        if port:
    30	            args += ['-p', str(port)]
    31	        args += [dbname]
    32	
    33	        sigint_handler = signal.getsignal(signal.SIGINT)
    34	        subprocess_env = os.environ.copy()
    35	        if passwd:
    36	            subprocess_env['PGPASSWORD'] = str(passwd)
    37	        if sslmode:
    38	            subprocess_env['PGSSLMODE'] = str(sslmode)
    39	        if sslrootcert:
    40	            subprocess_env['PGSSLROOTCERT'] = str(sslrootcert)
    41	        if sslcert:
    42	            subprocess_env['PGSSLCERT'] = str(sslcert)
    43	        if sslkey:
    44	            subprocess_env['PGSSLKEY'] = str(sslkey)
    45	        try:
    46	            # Allow SIGINT to pass to psql to abort queries.
    47	            signal.signal(signal.SIGINT, signal.SIG_IGN)
    48	            subprocess.run(args, check=True, env=subprocess_env)
    49	        finally:
    50	            # Restore the original SIGINT handler.
    51	            signal.signal(signal.SIGINT, sigint_handler)
    52	
    53	    def runshell(self):
    54	        DatabaseClient.runshell_db(self.connection.get_connection_params())

---
     1	import os
     2	import signal
     3	import subprocess
     4	from unittest import mock
     5	
     6	from django.db.backends.postgresql.client import DatabaseClient
     7	from django.test import SimpleTestCase
     8	
     9	
    10	class PostgreSqlDbshellCommandTestCase(SimpleTestCase):
    11	
    12	    def _run_it(self, dbinfo):
    13	        """
    14	        That function invokes the runshell command, while mocking
    15	        subprocess.run(). It returns a 2-tuple with:
    16	        - The command line list
    17	        - The the value of the PGPASSWORD environment variable, or None.
    18	        """
    19	        def _mock_subprocess_run(*args, env=os.environ, **kwargs):
    20	            self.subprocess_args = list(*args)
    21	            self.pgpassword = env.get('PGPASSWORD')
    22	            return subprocess.CompletedProcess(self.subprocess_args, 0)
    23	        with mock.patch('subprocess.run', new=_mock_subprocess_run):
    24	            DatabaseClient.runshell_db(dbinfo)
    25	        return self.subprocess_args, self.pgpassword
    26	
    27	    def test_basic(self):
    28	        self.assertEqual(
    29	            self._run_it({
    30	                'database': 'dbname',
    31	                'user': 'someuser',
    32	                'password': 'somepassword',
    33	                'host': 'somehost',
    34	                'port': '444',
    35	            }), (
    36	                ['psql', '-U', 'someuser', '-h', 'somehost', '-p', '444', 'dbname'],
    37	                'somepassword',
    38	            )
    39	        )
    40	
    41	    def test_nopass(self):
    42	        self.assertEqual(
    43	            self._run_it({
    44	                'database': 'dbname',
    45	                'user': 'someuser',
    46	                'host': 'somehost',
    47	                'port': '444',
    48	            }), (
    49	                ['psql', '-U', 'someuser', '-h', 'somehost', '-p', '444', 'dbname'],
    50	                None,
    51	            )
    52	        )
    53	
    54	    def test_column(self):
    55	        self.assertEqual(
    56	            self._run_it({
    57	                'database': 'dbname',
    58	                'user': 'some:user',
    59	                'password': 'some:password',
    60	                'host': '::1',
    61	                'port': '444',
    62	            }), (
    63	                ['psql', '-U', 'some:user', '-h', '::1', '-p', '444', 'dbname'],
    64	                'some:password',
    65	            )
    66	        )
    67	
    68	    def test_accent(self):
    69	        username = 'rôle'
    70	        password = 'sésame'
    71	        self.assertEqual(
    72	            self._run_it({
    73	                'database': 'dbname',
    74	                'user': username,
    75	                'password': password,
    76	                'host': 'somehost',
    77	                'port': '444',
    78	            }), (
    79	                ['psql', '-U', username, '-h', 'somehost', '-p', '444', 'dbname'],
    80	                password,
    81	            )
    82	        )
    83	
    84	    def test_sigint_handler(self):
    85	        """SIGINT is ignored in Python and passed to psql to abort quries."""
    86	        def _mock_subprocess_run(*args, **kwargs):
    87	            handler = signal.getsignal(signal.SIGINT)
    88	            self.assertEqual(handler, signal.SIG_IGN)
    89	
    90	        sigint_handler = signal.getsignal(signal.SIGINT)
    91	        # The default handler isn't SIG_IGN.
    92	        self.assertNotEqual(sigint_handler, signal.SIG_IGN)
    93	        with mock.patch('subprocess.run', new=_mock_subprocess_run):
    94	            DatabaseClient.runshell_db({})
    95	        # dbshell restores the original handler.
    96	        self.assertEqual(sigint_handler, signal.getsignal(signal.SIGINT))