Refs #32061 -- Unified DatabaseClient.runshell() in db backends.

4 files changed, 165 insertions, 140 deletions

diff --git a/tests/dbshell/test_mysql.py b/tests/dbshell/test_mysql.py
index e09644962b..374c466059 100644
--- a/tests/dbshell/test_mysql.py
+++ b/tests/dbshell/test_mysql.py
@@ -3,32 +3,52 @@ from django.test import SimpleTestCase
 
 
 class MySqlDbshellCommandTestCase(SimpleTestCase):
+    def settings_to_cmd_args_env(self, settings_dict, parameters=None):
+        if parameters is None:
+            parameters = []
+        return DatabaseClient.settings_to_cmd_args_env(settings_dict, parameters)
 
     def test_fails_with_keyerror_on_incomplete_config(self):
         with self.assertRaises(KeyError):
-            self.get_command_line_arguments({})
+            self.settings_to_cmd_args_env({})
 
     def test_basic_params_specified_in_settings(self):
+        expected_args = [
+            'mysql',
+            '--user=someuser',
+            '--password=somepassword',
+            '--host=somehost',
+            '--port=444',
+            'somedbname',
+        ]
+        expected_env = None
         self.assertEqual(
-            ['mysql', '--user=someuser', '--password=somepassword',
-             '--host=somehost', '--port=444', 'somedbname'],
-            self.get_command_line_arguments({
+            self.settings_to_cmd_args_env({
                 'NAME': 'somedbname',
                 'USER': 'someuser',
                 'PASSWORD': 'somepassword',
                 'HOST': 'somehost',
                 'PORT': 444,
                 'OPTIONS': {},
-            }))
+            }),
+            (expected_args, expected_env),
+        )
 
     def test_options_override_settings_proper_values(self):
         settings_port = 444
         options_port = 555
         self.assertNotEqual(settings_port, options_port, 'test pre-req')
+        expected_args = [
+            'mysql',
+            '--user=optionuser',
+            '--password=optionpassword',
+            '--host=optionhost',
+            '--port=%s' % options_port,
+            'optiondbname',
+        ]
+        expected_env = None
         self.assertEqual(
-            ['mysql', '--user=optionuser', '--password=optionpassword',
-             '--host=optionhost', '--port={}'.format(options_port), 'optiondbname'],
-            self.get_command_line_arguments({
+            self.settings_to_cmd_args_env({
                 'NAME': 'settingdbname',
                 'USER': 'settinguser',
                 'PASSWORD': 'settingpassword',
@@ -41,15 +61,22 @@ class MySqlDbshellCommandTestCase(SimpleTestCase):
                     'host': 'optionhost',
                     'port': options_port,
                 },
-            }))
+            }),
+            (expected_args, expected_env),
+        )
 
     def test_options_password(self):
+        expected_args = [
+            'mysql',
+            '--user=someuser',
+            '--password=optionpassword',
+            '--host=somehost',
+            '--port=444',
+            'somedbname',
+        ]
+        expected_env = None
         self.assertEqual(
-            [
-                'mysql', '--user=someuser', '--password=optionpassword',
-                '--host=somehost', '--port=444', 'somedbname',
-            ],
-            self.get_command_line_arguments({
+            self.settings_to_cmd_args_env({
                 'NAME': 'somedbname',
                 'USER': 'someuser',
                 'PASSWORD': 'settingpassword',
@@ -57,16 +84,22 @@ class MySqlDbshellCommandTestCase(SimpleTestCase):
                 'PORT': 444,
                 'OPTIONS': {'password': 'optionpassword'},
             }),
+            (expected_args, expected_env),
         )
 
     def test_options_charset(self):
+        expected_args = [
+            'mysql',
+            '--user=someuser',
+            '--password=somepassword',
+            '--host=somehost',
+            '--port=444',
+            '--default-character-set=utf8',
+            'somedbname',
+        ]
+        expected_env = None
         self.assertEqual(
-            [
-                'mysql', '--user=someuser', '--password=somepassword',
-                '--host=somehost', '--port=444',
-                '--default-character-set=utf8', 'somedbname',
-            ],
-            self.get_command_line_arguments({
+            self.settings_to_cmd_args_env({
                 'NAME': 'somedbname',
                 'USER': 'someuser',
                 'PASSWORD': 'somepassword',
@@ -74,27 +107,45 @@ class MySqlDbshellCommandTestCase(SimpleTestCase):
                 'PORT': 444,
                 'OPTIONS': {'charset': 'utf8'},
             }),
+            (expected_args, expected_env),
         )
 
     def test_can_connect_using_sockets(self):
+        expected_args = [
+            'mysql',
+            '--user=someuser',
+            '--password=somepassword',
+            '--socket=/path/to/mysql.socket.file',
+            'somedbname',
+        ]
+        expected_env = None
         self.assertEqual(
-            ['mysql', '--user=someuser', '--password=somepassword',
-             '--socket=/path/to/mysql.socket.file', 'somedbname'],
-            self.get_command_line_arguments({
+            self.settings_to_cmd_args_env({
                 'NAME': 'somedbname',
                 'USER': 'someuser',
                 'PASSWORD': 'somepassword',
                 'HOST': '/path/to/mysql.socket.file',
                 'PORT': None,
                 'OPTIONS': {},
-            }))
+            }),
+            (expected_args, expected_env),
+        )
 
     def test_ssl_certificate_is_added(self):
+        expected_args = [
+            'mysql',
+            '--user=someuser',
+            '--password=somepassword',
+            '--host=somehost',
+            '--port=444',
+            '--ssl-ca=sslca',
+            '--ssl-cert=sslcert',
+            '--ssl-key=sslkey',
+            'somedbname',
+        ]
+        expected_env = None
         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({
+            self.settings_to_cmd_args_env({
                 'NAME': 'somedbname',
                 'USER': 'someuser',
                 'PASSWORD': 'somepassword',
@@ -107,12 +158,13 @@ class MySqlDbshellCommandTestCase(SimpleTestCase):
                         'key': 'sslkey',
                     },
                 },
-            }))
+            }),
+            (expected_args, expected_env),
+        )
 
     def test_parameters(self):
         self.assertEqual(
-            ['mysql', 'somedbname', '--help'],
-            self.get_command_line_arguments(
+            self.settings_to_cmd_args_env(
                 {
                     'NAME': 'somedbname',
                     'USER': None,
@@ -123,9 +175,5 @@ class MySqlDbshellCommandTestCase(SimpleTestCase):
                 },
                 ['--help'],
             ),
+            (['mysql', 'somedbname', '--help'], None),
         )
-
-    def get_command_line_arguments(self, connection_settings, parameters=None):
-        if parameters is None:
-            parameters = []
-        return DatabaseClient.settings_to_cmd_args(connection_settings, parameters)
diff --git a/tests/dbshell/test_oracle.py b/tests/dbshell/test_oracle.py
index 41fbc07455..34e96fb09b 100644
--- a/tests/dbshell/test_oracle.py
+++ b/tests/dbshell/test_oracle.py
@@ -1,4 +1,3 @@
-from subprocess import CompletedProcess
 from unittest import mock, skipUnless
 
 from django.db import connection
@@ -6,37 +5,48 @@ from django.db.backends.oracle.client import DatabaseClient
 from django.test import SimpleTestCase
 
 
-@skipUnless(connection.vendor == 'oracle', 'Oracle tests')
+@skipUnless(connection.vendor == 'oracle', 'Requires cx_Oracle to be installed')
 class OracleDbshellTests(SimpleTestCase):
-    def _run_dbshell(self, rlwrap=False, parameters=None):
-        """Run runshell command and capture its arguments."""
-        def _mock_subprocess_run(*args, **kwargs):
-            self.subprocess_args = list(*args)
-            return CompletedProcess(self.subprocess_args, 0)
-
+    def settings_to_cmd_args_env(self, settings_dict, parameters=None, rlwrap=False):
         if parameters is None:
             parameters = []
-        client = DatabaseClient(connection)
-        self.subprocess_args = None
-        with mock.patch('subprocess.run', new=_mock_subprocess_run):
-            with mock.patch('shutil.which', return_value='/usr/bin/rlwrap' if rlwrap else None):
-                client.runshell(parameters)
-        return self.subprocess_args
+        with mock.patch('shutil.which', return_value='/usr/bin/rlwrap' if rlwrap else None):
+            return DatabaseClient.settings_to_cmd_args_env(settings_dict, parameters)
 
     def test_without_rlwrap(self):
+        expected_args = [
+            'sqlplus',
+            '-L',
+            connection.client.connect_string(connection.settings_dict),
+        ]
         self.assertEqual(
-            self._run_dbshell(rlwrap=False),
-            ['sqlplus', '-L', connection._connect_string()],
+            self.settings_to_cmd_args_env(connection.settings_dict, rlwrap=False),
+            (expected_args, None),
         )
 
     def test_with_rlwrap(self):
+        expected_args = [
+            '/usr/bin/rlwrap',
+            'sqlplus',
+            '-L',
+            connection.client.connect_string(connection.settings_dict),
+        ]
         self.assertEqual(
-            self._run_dbshell(rlwrap=True),
-            ['/usr/bin/rlwrap', 'sqlplus', '-L', connection._connect_string()],
+            self.settings_to_cmd_args_env(connection.settings_dict, rlwrap=True),
+            (expected_args, None),
         )
 
     def test_parameters(self):
+        expected_args = [
+            'sqlplus',
+            '-L',
+            connection.client.connect_string(connection.settings_dict),
+            '-HELP',
+        ]
         self.assertEqual(
-            self._run_dbshell(parameters=['-HELP']),
-            ['sqlplus', '-L', connection._connect_string(), '-HELP'],
+            self.settings_to_cmd_args_env(
+                connection.settings_dict,
+                parameters=['-HELP'],
+            ),
+            (expected_args, None),
         )
diff --git a/tests/dbshell/test_postgresql.py b/tests/dbshell/test_postgresql.py
index 6de60eaef2..aad9692ecb 100644
--- a/tests/dbshell/test_postgresql.py
+++ b/tests/dbshell/test_postgresql.py
@@ -1,41 +1,25 @@
-import os
 import signal
-import subprocess
-from unittest import mock
+from unittest import mock, skipUnless
 
+from django.db import connection
 from django.db.backends.postgresql.client import DatabaseClient
 from django.test import SimpleTestCase
 
 
 class PostgreSqlDbshellCommandTestCase(SimpleTestCase):
-
-    def _run_it(self, dbinfo, parameters=None):
-        """
-        That function invokes the runshell command, while mocking
-        subprocess.run(). It returns a 2-tuple with:
-        - The command line list
-        - The dictionary of PG* environment variables, or {}.
-        """
-        def _mock_subprocess_run(*args, env=os.environ, **kwargs):
-            self.subprocess_args = list(*args)
-            # PostgreSQL environment variables.
-            self.pg_env = {key: env[key] for key in env if key.startswith('PG')}
-            return subprocess.CompletedProcess(self.subprocess_args, 0)
-
+    def settings_to_cmd_args_env(self, settings_dict, parameters=None):
         if parameters is None:
             parameters = []
-        with mock.patch('subprocess.run', new=_mock_subprocess_run):
-            DatabaseClient.runshell_db(dbinfo, parameters)
-        return self.subprocess_args, self.pg_env
+        return DatabaseClient.settings_to_cmd_args_env(settings_dict, parameters)
 
     def test_basic(self):
         self.assertEqual(
-            self._run_it({
-                'database': 'dbname',
-                'user': 'someuser',
-                'password': 'somepassword',
-                'host': 'somehost',
-                'port': '444',
+            self.settings_to_cmd_args_env({
+                'NAME': 'dbname',
+                'USER': 'someuser',
+                'PASSWORD': 'somepassword',
+                'HOST': 'somehost',
+                'PORT': '444',
             }), (
                 ['psql', '-U', 'someuser', '-h', 'somehost', '-p', '444', 'dbname'],
                 {'PGPASSWORD': 'somepassword'},
@@ -44,11 +28,11 @@ class PostgreSqlDbshellCommandTestCase(SimpleTestCase):
 
     def test_nopass(self):
         self.assertEqual(
-            self._run_it({
-                'database': 'dbname',
-                'user': 'someuser',
-                'host': 'somehost',
-                'port': '444',
+            self.settings_to_cmd_args_env({
+                'NAME': 'dbname',
+                'USER': 'someuser',
+                'HOST': 'somehost',
+                'PORT': '444',
             }), (
                 ['psql', '-U', 'someuser', '-h', 'somehost', '-p', '444', 'dbname'],
                 {},
@@ -57,15 +41,17 @@ class PostgreSqlDbshellCommandTestCase(SimpleTestCase):
 
     def test_ssl_certificate(self):
         self.assertEqual(
-            self._run_it({
-                'database': 'dbname',
-                'user': 'someuser',
-                'host': 'somehost',
-                'port': '444',
-                'sslmode': 'verify-ca',
-                'sslrootcert': 'root.crt',
-                'sslcert': 'client.crt',
-                'sslkey': 'client.key',
+            self.settings_to_cmd_args_env({
+                'NAME': 'dbname',
+                'USER': 'someuser',
+                'HOST': 'somehost',
+                'PORT': '444',
+                'OPTIONS': {
+                    'sslmode': 'verify-ca',
+                    'sslrootcert': 'root.crt',
+                    'sslcert': 'client.crt',
+                    'sslkey': 'client.key',
+                },
             }), (
                 ['psql', '-U', 'someuser', '-h', 'somehost', '-p', '444', 'dbname'],
                 {
@@ -79,12 +65,12 @@ class PostgreSqlDbshellCommandTestCase(SimpleTestCase):
 
     def test_column(self):
         self.assertEqual(
-            self._run_it({
-                'database': 'dbname',
-                'user': 'some:user',
-                'password': 'some:password',
-                'host': '::1',
-                'port': '444',
+            self.settings_to_cmd_args_env({
+                'NAME': 'dbname',
+                'USER': 'some:user',
+                'PASSWORD': 'some:password',
+                'HOST': '::1',
+                'PORT': '444',
             }), (
                 ['psql', '-U', 'some:user', '-h', '::1', '-p', '444', 'dbname'],
                 {'PGPASSWORD': 'some:password'},
@@ -95,12 +81,12 @@ class PostgreSqlDbshellCommandTestCase(SimpleTestCase):
         username = 'rôle'
         password = 'sésame'
         self.assertEqual(
-            self._run_it({
-                'database': 'dbname',
-                'user': username,
-                'password': password,
-                'host': 'somehost',
-                'port': '444',
+            self.settings_to_cmd_args_env({
+                'NAME': 'dbname',
+                'USER': username,
+                'PASSWORD': password,
+                'HOST': 'somehost',
+                'PORT': '444',
             }), (
                 ['psql', '-U', username, '-h', 'somehost', '-p', '444', 'dbname'],
                 {'PGPASSWORD': password},
@@ -109,10 +95,11 @@ class PostgreSqlDbshellCommandTestCase(SimpleTestCase):
 
     def test_parameters(self):
         self.assertEqual(
-            self._run_it({'database': 'dbname'}, ['--help']),
+            self.settings_to_cmd_args_env({'NAME': 'dbname'}, ['--help']),
             (['psql', 'dbname', '--help'], {}),
         )
 
+    @skipUnless(connection.vendor == 'postgresql', 'Requires a PostgreSQL connection')
     def test_sigint_handler(self):
         """SIGINT is ignored in Python and passed to psql to abort queries."""
         def _mock_subprocess_run(*args, **kwargs):
@@ -123,6 +110,6 @@ class PostgreSqlDbshellCommandTestCase(SimpleTestCase):
         # 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({}, [])
+            connection.client.runshell([])
         # dbshell restores the original handler.
         self.assertEqual(sigint_handler, signal.getsignal(signal.SIGINT))
diff --git a/tests/dbshell/test_sqlite.py b/tests/dbshell/test_sqlite.py
index c3b2b1e28d..570230f62d 100644
--- a/tests/dbshell/test_sqlite.py
+++ b/tests/dbshell/test_sqlite.py
@@ -1,43 +1,23 @@
 from pathlib import Path
-from subprocess import CompletedProcess
-from unittest import mock, skipUnless
 
-from django.db import connection
 from django.db.backends.sqlite3.client import DatabaseClient
 from django.test import SimpleTestCase
 
 
-@skipUnless(connection.vendor == 'sqlite', 'SQLite tests.')
 class SqliteDbshellCommandTestCase(SimpleTestCase):
-    def _run_dbshell(self, parameters=None):
-        """Run runshell command and capture its arguments."""
-        def _mock_subprocess_run(*args, **kwargs):
-            self.subprocess_args = list(*args)
-            return CompletedProcess(self.subprocess_args, 0)
-
+    def settings_to_cmd_args_env(self, settings_dict, parameters=None):
         if parameters is None:
             parameters = []
-        client = DatabaseClient(connection)
-        with mock.patch('subprocess.run', new=_mock_subprocess_run):
-            client.runshell(parameters)
-        return self.subprocess_args
+        return DatabaseClient.settings_to_cmd_args_env(settings_dict, parameters)
 
     def test_path_name(self):
-        with mock.patch.dict(
-            connection.settings_dict,
-            {'NAME': Path('test.db.sqlite3')},
-        ):
-            self.assertEqual(
-                self._run_dbshell(),
-                ['sqlite3', 'test.db.sqlite3'],
-            )
+        self.assertEqual(
+            self.settings_to_cmd_args_env({'NAME': Path('test.db.sqlite3')}),
+            (['sqlite3', 'test.db.sqlite3'], None),
+        )
 
     def test_parameters(self):
-        with mock.patch.dict(
-            connection.settings_dict,
-            {'NAME': Path('test.db.sqlite3')},
-        ):
-            self.assertEqual(
-                self._run_dbshell(['-help']),
-                ['sqlite3', 'test.db.sqlite3', '-help'],
-            )
+        self.assertEqual(
+            self.settings_to_cmd_args_env({'NAME': 'test.db.sqlite3'}, ['-help']),
+            (['sqlite3', 'test.db.sqlite3', '-help'], None),
+        )