1
0
mirror of https://github.com/django/django.git synced 2025-06-05 03:29:12 +00:00

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.
This commit is contained in:
Mariusz Felisiak 2024-09-18 21:18:54 +05:00 committed by Sarah Boyce
parent 30e0a43937
commit 1823a80113
2 changed files with 67 additions and 8 deletions

View File

@ -74,14 +74,27 @@ 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:
with subprocess.Popen(
with (
subprocess.Popen(
dump_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=dump_env
) as dump_proc,
subprocess.Popen(
load_cmd,
stdin=dump_proc.stdout,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
env=load_env,
):
# Allow dump_proc to receive a SIGPIPE if the load process exits.
dump_proc.stdout.close()
) as load_proc,
):
# Allow dump_proc to receive a SIGPIPE if the load process exits.
dump_proc.stdout.close()
dump_err = dump_proc.stderr.read().decode(errors="replace")
load_err = load_proc.stderr.read().decode(errors="replace")
if dump_proc.returncode != 0:
self.log(
f"Got an error on mysqldump when cloning the test database: {dump_err}"
)
sys.exit(dump_proc.returncode)
if load_proc.returncode != 0:
self.log(f"Got an error cloning the test database: {load_err}")
sys.exit(load_proc.returncode)

View File

@ -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())