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