simulate test database creation raising database exists systemexit is raised if the user answers no to the prompt asking if it s okay to delete the test database database exists shouldn t appear when keepdb is on simulate test database creation raising unexpected error simulate test database creation raising database exists systemexit is raised if the user answers no to the prompt asking if it s okay to delete the test database database exists shouldn t appear when keepdb is on simulate test database creation raising unexpected error
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):
        
        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):
                    
                    
                    creation._create_test_db(
                        verbosity=0, autoclobber=False, keepdb=False
                    )
            
            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):
        
        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