From c263ac2d7c0b6e45e8eb5cd56410b01c730a93d3 Mon Sep 17 00:00:00 2001 From: Faakhir30 Date: Wed, 18 Sep 2024 21:18:54 +0500 Subject: [PATCH] refs #33537: Added ability to exit if mysqldump fails. --- django/db/backends/mysql/creation.py | 41 ++++++-- tests/backends/mysql/test_creation.py | 143 ++++++++++++++++++++++---- 2 files changed, 156 insertions(+), 28 deletions(-) diff --git a/django/db/backends/mysql/creation.py b/django/db/backends/mysql/creation.py index a060f41d18..fb2d9b5010 100644 --- a/django/db/backends/mysql/creation.py +++ b/django/db/backends/mysql/creation.py @@ -74,14 +74,35 @@ class DatabaseCreation(BaseDatabaseCreation): load_cmd = cmd_args load_cmd[-1] = target_database_name - with subprocess.Popen( - dump_cmd, stdout=subprocess.PIPE, env=dump_env - ) as dump_proc: + try: with subprocess.Popen( - load_cmd, - stdin=dump_proc.stdout, - stdout=subprocess.DEVNULL, - env=load_env, - ): - # Allow dump_proc to receive a SIGPIPE if the load process exits. - dump_proc.stdout.close() + dump_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=dump_env + ) as dump_proc: + with subprocess.Popen( + load_cmd, + stdin=dump_proc.stdout, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=load_env, + ) as load_proc: + dump_proc.stdout.close() + load_out, load_err = load_proc.communicate() + dump_err = dump_proc.stderr.read().decode() + + if dump_proc.returncode != 0: + self.log( + "Got a fatal error cloning the test database (dump): %s" % dump_err + ) + sys.exit(2) + + if load_proc.returncode != 0: + self.log( + "Got a fatal error cloning the test database (load): %s" % load_err + ) + sys.exit(2) + + except Exception as e: + self.log("An exception occurred while cloning the database: %s" % e) + raise + + self.log("Database cloning process finished.") diff --git a/tests/backends/mysql/test_creation.py b/tests/backends/mysql/test_creation.py index 151d00ff3f..abef8aa898 100644 --- a/tests/backends/mysql/test_creation.py +++ b/tests/backends/mysql/test_creation.py @@ -56,7 +56,20 @@ class DatabaseCreationTests(SimpleTestCase): creation._clone_test_db("suffix", verbosity=0, keepdb=True) _clone_db.assert_not_called() - def test_clone_test_db_options_ordering(self): + @mock.patch("subprocess.Popen") + def test_clone_test_db_unexpected_error(self, mocked_popen): + creation = DatabaseCreation(connection) + mocked_proc = mock.Mock() + mocked_proc.communicate.return_value = (b"stdout", b"stderr") + mocked_popen.return_value.__enter__.return_value = mocked_proc + + with self.assertRaises(SystemExit): + creation._clone_db("source_db", "target_db") + + @mock.patch("sys.stdout", new_callable=StringIO) + @mock.patch("sys.stderr", new_callable=StringIO) + @mock.patch("subprocess.Popen") + def test_clone_test_db_options_ordering(self, mocked_popen, *mocked_objects): creation = DatabaseCreation(connection) try: saved_settings = connection.settings_dict @@ -71,22 +84,116 @@ class DatabaseCreationTests(SimpleTestCase): "read_default_file": "my.cnf", }, } - with mock.patch.object(subprocess, "Popen") as mocked_popen: - creation._clone_db("source_db", "target_db") - mocked_popen.assert_has_calls( - [ - mock.call( - [ - "mysqldump", - "--defaults-file=my.cnf", - "--routines", - "--events", - "source_db", - ], - stdout=subprocess.PIPE, - env=None, - ), - ] - ) + mock_proc = mock.Mock() + mock_proc.communicate.return_value = (b"", b"") + mock_proc.returncode = 0 + mocked_popen.return_value.__enter__.return_value = mock_proc + + creation._clone_db("source_db", "target_db") + mocked_popen.assert_has_calls( + [ + mock.call( + [ + "mysqldump", + "--defaults-file=my.cnf", + "--routines", + "--events", + "source_db", + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=None, + ), + ] + ) + finally: + connection.settings_dict = saved_settings + + @mock.patch("sys.stdout", new_callable=StringIO) + @mock.patch("sys.stderr", new_callable=StringIO) + @mock.patch("subprocess.Popen") + def test_clone_test_db_dump_error(self, mocked_popen, *mocked_objects): + creation = DatabaseCreation(connection) + try: + saved_settings = connection.settings_dict + connection.settings_dict = { + "NAME": "source_db", + "USER": "", + "PASSWORD": "", + "PORT": "", + "HOST": "", + "ENGINE": "django.db.backends.mysql", + "OPTIONS": { + "read_default_file": "my.cnf", + }, + } + mock_proc = mock.Mock() + mock_proc.communicate.return_value = (b"", b"Dump error") + mock_proc.returncode = 1 # Simulate dump failure + mocked_popen.return_value.__enter__.return_value = mock_proc + + with self.assertRaises(SystemExit) as cm: + creation._clone_db("source_db", "target_db") + self.assertEqual(cm.exception.code, 2) + + mocked_popen.assert_called_with( + [ + "mysqldump", + "--defaults-file=my.cnf", + "--routines", + "--events", + "source_db", + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=None, + ) + finally: + connection.settings_dict = saved_settings + + @mock.patch("sys.stdout", new_callable=StringIO) + @mock.patch("sys.stderr", new_callable=StringIO) + @mock.patch("subprocess.Popen") + def test_clone_test_db_load_error(self, mocked_popen, *mocked_objects): + creation = DatabaseCreation(connection) + try: + saved_settings = connection.settings_dict + connection.settings_dict = { + "NAME": "source_db", + "USER": "", + "PASSWORD": "", + "PORT": "", + "HOST": "", + "ENGINE": "django.db.backends.mysql", + "OPTIONS": { + "read_default_file": "my.cnf", + }, + } + mock_dump_proc = mock.Mock() + mock_dump_proc.communicate.return_value = (b"", b"") + mock_dump_proc.returncode = 0 # Simulate successful dump + + mock_load_proc = mock.Mock() + mock_load_proc.communicate.return_value = (b"", b"Load error") + mock_load_proc.returncode = 1 # Simulate load failure + + mocked_popen.side_effect = [mock_dump_proc, mock_load_proc] + + with self.assertRaises(SystemExit) as cm: + creation._clone_db("source_db", "target_db") + self.assertEqual(cm.exception.code, 2) + + mocked_popen.assert_called_with( + [ + "mysqldump", + "--defaults-file=my.cnf", + "--routines", + "--events", + "source_db", + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=None, + ) finally: connection.settings_dict = saved_settings