86 lines
3.8 KiB
Python
86 lines
3.8 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
|