mirror of
https://github.com/django/django.git
synced 2025-10-31 01:25:32 +00:00
93 lines
4.0 KiB
Python
93 lines
4.0 KiB
Python
import subprocess
|
|
import unittest
|
|
from io import 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
|
|
|
|
|
|
@unittest.skipUnless(connection.vendor == "mysql", "MySQL tests")
|
|
class DatabaseCreationTests(SimpleTestCase):
|
|
def _execute_raise_database_exists(self, cursor, parameters, keepdb=False):
|
|
raise DatabaseError(
|
|
1007, "Can't create database '%s'; database exists" % parameters["dbname"]
|
|
)
|
|
|
|
def _execute_raise_access_denied(self, cursor, parameters, keepdb=False):
|
|
raise DatabaseError(1044, "Access denied for user")
|
|
|
|
def patch_test_db_creation(self, execute_create_test_db):
|
|
return mock.patch.object(
|
|
BaseDatabaseCreation, "_execute_create_test_db", execute_create_test_db
|
|
)
|
|
|
|
@mock.patch("sys.stdout", new_callable=StringIO)
|
|
@mock.patch("sys.stderr", new_callable=StringIO)
|
|
def test_create_test_db_database_exists(self, *mocked_objects):
|
|
# Simulate test database creation raising "database exists"
|
|
creation = DatabaseCreation(connection)
|
|
with self.patch_test_db_creation(self._execute_raise_database_exists):
|
|
with mock.patch("builtins.input", return_value="no"):
|
|
with self.assertRaises(SystemExit):
|
|
# SystemExit is raised if the user answers "no" to the
|
|
# prompt asking if it's okay to delete the test database.
|
|
creation._create_test_db(
|
|
verbosity=0, autoclobber=False, keepdb=False
|
|
)
|
|
# "Database exists" shouldn't appear when keepdb is on
|
|
creation._create_test_db(verbosity=0, autoclobber=False, keepdb=True)
|
|
|
|
@mock.patch("sys.stdout", new_callable=StringIO)
|
|
@mock.patch("sys.stderr", new_callable=StringIO)
|
|
def test_create_test_db_unexpected_error(self, *mocked_objects):
|
|
# Simulate test database creation raising unexpected error
|
|
creation = DatabaseCreation(connection)
|
|
with self.patch_test_db_creation(self._execute_raise_access_denied):
|
|
with self.assertRaises(SystemExit):
|
|
creation._create_test_db(verbosity=0, autoclobber=False, keepdb=False)
|
|
|
|
def test_clone_test_db_database_exists(self):
|
|
creation = DatabaseCreation(connection)
|
|
with self.patch_test_db_creation(self._execute_raise_database_exists):
|
|
with mock.patch.object(DatabaseCreation, "_clone_db") as _clone_db:
|
|
creation._clone_test_db("suffix", verbosity=0, keepdb=True)
|
|
_clone_db.assert_not_called()
|
|
|
|
def test_clone_test_db_options_ordering(self):
|
|
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",
|
|
},
|
|
}
|
|
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,
|
|
),
|
|
]
|
|
)
|
|
finally:
|
|
connection.settings_dict = saved_settings
|