refs #33537: Added ability to exit if mysqldump fails.

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(
+        try:
-            dump_cmd, stdout=subprocess.PIPE, env=dump_env
-        ) as dump_proc:
             with subprocess.Popen(
-                load_cmd,
+                dump_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=dump_env
-                stdin=dump_proc.stdout,
+            ) as dump_proc:
-                stdout=subprocess.DEVNULL,
+                with subprocess.Popen(
-                env=load_env,
+                    load_cmd,
-            ):
+                    stdin=dump_proc.stdout,
-                # Allow dump_proc to receive a SIGPIPE if the load process exits.
+                    stdout=subprocess.PIPE,
-                dump_proc.stdout.close()
+                    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.")

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:
+            mock_proc = mock.Mock()
-                creation._clone_db("source_db", "target_db")
+            mock_proc.communicate.return_value = (b"", b"")
-                mocked_popen.assert_has_calls(
+            mock_proc.returncode = 0
-                    [
+            mocked_popen.return_value.__enter__.return_value = mock_proc
-                        mock.call(
+
-                            [
+            creation._clone_db("source_db", "target_db")
-                                "mysqldump",
+            mocked_popen.assert_has_calls(
-                                "--defaults-file=my.cnf",
+                [
-                                "--routines",
+                    mock.call(
-                                "--events",
+                        [
-                                "source_db",
+                            "mysqldump",
-                            ],
+                            "--defaults-file=my.cnf",
-                            stdout=subprocess.PIPE,
+                            "--routines",
-                            env=None,
+                            "--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