Fixed #33537 -- Made test database cloning on MySQL reraise unexpected errors.

Thanks Faakhir Zahid and Stephen Finucane for the initial patch.

Thanks Simon Charette for the review.

1 files changed, 47 insertions, 1 deletions

diff --git a/tests/backends/mysql/test_creation.py b/tests/backends/mysql/test_creation.py
index 151d00ff3f..38765029dd 100644
--- a/tests/backends/mysql/test_creation.py
+++ b/tests/backends/mysql/test_creation.py
@@ -1,12 +1,13 @@
 import subprocess
 import unittest
-from io import StringIO
+from io import BytesIO, StringIO
 from unittest import mock
 
 from django.db import DatabaseError, connection
 from django.db.backends.base.creation import BaseDatabaseCreation
 from django.db.backends.mysql.creation import DatabaseCreation
 from django.test import SimpleTestCase
+from django.test.utils import captured_stderr
 
 
 @unittest.skipUnless(connection.vendor == "mysql", "MySQL tests")
@@ -58,6 +59,8 @@ class DatabaseCreationTests(SimpleTestCase):
 
     def test_clone_test_db_options_ordering(self):
         creation = DatabaseCreation(connection)
+        mock_subprocess_call = mock.MagicMock()
+        mock_subprocess_call.returncode = 0
         try:
             saved_settings = connection.settings_dict
             connection.settings_dict = {
@@ -72,6 +75,7 @@ class DatabaseCreationTests(SimpleTestCase):
                 },
             }
             with mock.patch.object(subprocess, "Popen") as mocked_popen:
+                mocked_popen.return_value.__enter__.return_value = mock_subprocess_call
                 creation._clone_db("source_db", "target_db")
                 mocked_popen.assert_has_calls(
                     [
@@ -84,9 +88,51 @@ class DatabaseCreationTests(SimpleTestCase):
                                 "source_db",
                             ],
                             stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE,
                             env=None,
                         ),
                     ]
                 )
         finally:
             connection.settings_dict = saved_settings
+
+    def test_clone_test_db_subprocess_mysqldump_error(self):
+        creation = DatabaseCreation(connection)
+        mock_subprocess_call = mock.MagicMock()
+        mock_subprocess_call.returncode = 0
+        # Simulate mysqldump in test database cloning raises an error.
+        msg = "Couldn't execute 'SELECT ...'"
+        mock_subprocess_call_error = mock.MagicMock()
+        mock_subprocess_call_error.returncode = 2
+        mock_subprocess_call_error.stderr = BytesIO(msg.encode())
+        with mock.patch.object(subprocess, "Popen") as mocked_popen:
+            mocked_popen.return_value.__enter__.side_effect = [
+                mock_subprocess_call_error,  # mysqldump mock
+                mock_subprocess_call,  # load mock
+            ]
+            with captured_stderr() as err, self.assertRaises(SystemExit) as cm:
+                creation._clone_db("source_db", "target_db")
+            self.assertEqual(cm.exception.code, 2)
+        self.assertIn(
+            f"Got an error on mysqldump when cloning the test database: {msg}",
+            err.getvalue(),
+        )
+
+    def test_clone_test_db_subprocess_mysql_error(self):
+        creation = DatabaseCreation(connection)
+        mock_subprocess_call = mock.MagicMock()
+        mock_subprocess_call.returncode = 0
+        # Simulate load in test database cloning raises an error.
+        msg = "Some error"
+        mock_subprocess_call_error = mock.MagicMock()
+        mock_subprocess_call_error.returncode = 3
+        mock_subprocess_call_error.stderr = BytesIO(msg.encode())
+        with mock.patch.object(subprocess, "Popen") as mocked_popen:
+            mocked_popen.return_value.__enter__.side_effect = [
+                mock_subprocess_call,  # mysqldump mock
+                mock_subprocess_call_error,  # load mock
+            ]
+            with captured_stderr() as err, self.assertRaises(SystemExit) as cm:
+                creation._clone_db("source_db", "target_db")
+            self.assertEqual(cm.exception.code, 3)
+        self.assertIn(f"Got an error cloning the test database: {msg}", err.getvalue())