import json
import ntpath
import os
import posixpath
import random
import string

from parallels.core import MigrationError
from parallels.core import messages, MigrationNoRepeatError
from parallels.core.logging import get_logger
from parallels.core.migrator_config import MSSQLCopyMethod
from parallels.core.runners.base import BaseRunner
from parallels.core.runners.entities import MSSQLConnectionSettings, ExecutionOptions
from parallels.core.runners.exceptions.non_zero_exit_code import NonZeroExitCodeException
from parallels.core.runners.utils import transfer_file
from parallels.core.runners.windows.base import WindowsRunner
from parallels.core.utils import windows_utils, unix_utils
from parallels.core.utils.common import split_nonempty_strs, is_empty, add_dict_key_prefix, cached, safe_format
from parallels.core.utils.common.ip import resolve_all_safe
from parallels.core.utils.common.threading_utils import synchronized
from parallels.core.utils.database_server_type import DatabaseServerType
from parallels.core.utils.entity import Entity
from parallels.core.utils.file_utils import temp_filename
from parallels.core.utils.plesk_utils import get_windows_temp_dump_dir
from parallels.core.utils.unix_utils import move_file_secure
from parallels.core.utils.windows_utils import get_from_registry

logger = get_logger(__name__)


class DatabaseInfo(object):
    """Information about database to copy"""

    def __init__(
        self, source_database_server, target_database_server, subscription_name,
        source_database_name, target_database_name
    ):
        """
        :type source_database_server: parallels.core.connections.database_servers.base.DatabaseServer
        :type target_database_server: parallels.core.connections.database_servers.base.DatabaseServer
        :type subscription_name: str | unicode
        :type source_database_name: str | unicode
        :type target_database_name: str | unicode
        """
        self._source_database_server = source_database_server
        self._target_database_server = target_database_server
        self._subscription_name = subscription_name
        self._source_database_name = source_database_name
        self._target_database_name = target_database_name

    @property
    def source_database_server(self):
        """Source database server of database to copy

        :rtype: parallels.core.connections.database_servers.base.DatabaseServer
        """
        return self._source_database_server

    @property
    def target_database_server(self):
        """Target database server of database to copy

        :rtype: parallels.core.connections.database_servers.base.DatabaseServer
        """
        return self._target_database_server

    @property
    def subscription_name(self):
        """Name of subscription database belongs to

        :rtype: str | unicode
        """
        return self._subscription_name

    @property
    def source_database_name(self):
        """Name of database on the source server

        :rtype: str | unicode
        """
        return self._source_database_name

    @property
    def target_database_name(self):
        """Name of database on the target server

        :rtype: str | unicode
        """
        return self._target_database_name

    def __str__(self):
        return safe_format(
            messages.DATABASE_COPY_INFO_DESCRIPTION,
            subscription_name=self.subscription_name,
            source_database_name=self.source_database_name,
            source_server=self.source_database_server.description(),
            target_database_name=self.target_database_name,
            target_server=self.target_database_server.description()
        )


class DatabaseConnectionInfo(Entity):
    """Connection information to particular database - host, port, login, password, database name"""

    def __init__(self, host, port, login, password, db_name):
        """
        :type host: str | unicode
        :type port: str | unicode
        :type login: str | unicode
        :type password: str | unicode
        :type db_name: str | unicode
        """
        self._host = host
        self._port = port
        self._login = login
        self._password = password
        self._db_name = db_name

    @property
    def host(self):
        """
        :rtype: str | unicode
        """
        return self._host

    @property
    def port(self):
        """
        :rtype: str | unicode
        """
        return self._port

    @property
    def login(self):
        """
        :rtype: str | unicode
        """
        return self._login

    @property
    def password(self):
        """
        :rtype: str | unicode
        """
        return self._password

    @property
    def db_name(self):
        """
        :rtype: str | unicode
        """
        return self._db_name


class LinuxDatabaseTransferOptions(Entity):
    def __init__(self, owner, keyfile, dump_format):
        self._owner = owner
        self._keyfile = keyfile
        self._dump_format = dump_format

    @property
    def owner(self):
        return self._owner

    @property
    def keyfile(self):
        return self._keyfile

    @property
    def dump_format(self):
        return self._dump_format


def copy_db_content_linux(database_info, options):
    """
    :type database_info: parallels.core.utils.database_utils.DatabaseInfo
    :type options: parallels.core.utils.database_utils.LinuxDatabaseTransferOptions
    """
    source = database_info.source_database_server
    target = database_info.target_database_server

    logger.info(messages.LOG_COPY_DB_CONTENT.format(
        db_name=database_info.source_database_name, source=source.description(), target=target.description()
    ))

    if source.type() == DatabaseServerType.MYSQL:
        configuration = MysqlTransferConfiguration(
            database_info.source_database_name, database_info.target_database_name
        )
    elif source.type() == DatabaseServerType.POSTGRESQL:
        if options.dump_format == 'text':
            configuration = PostgresqlPlaintextTransferConfiguration(
                database_info.source_database_name, database_info.target_database_name
            )
            configuration.set_original_owner(options.owner)
        else:
            configuration = PostgresqlBinaryFormatTransferConfiguration(
                database_info.source_database_name, database_info.target_database_name
            )
    else:
        raise Exception(messages.CANNOT_TRANSFER_DATABASE_OF_UNSUPPORTED_TYPE % source.type())

    transfer_processor = LinuxDatabaseContentTransfer(configuration, options.keyfile)
    transfer_processor.transfer(source, target, database_info.source_database_name, database_info.target_database_name)


class DatabaseTransferConfiguration(object):
    """Configuration options and commands for database transfer."""

    def __init__(self, source_database_name, target_database_name):
        self.source_database_name = source_database_name
        self.target_database_name = target_database_name

    def get_dump_command(self, server, password_file, dump_filename):
        """Return a tuple of command template, command options, environment variables and standard input.

         Result is suitable for runner.sh().

        :type server: parallels.core.connections.database_servers.base.DatabaseServer
        :type password_file: str | unicode
        :type dump_filename: str | unicode
        :rtype: tuple
        """
        pass

    def get_restore_command(self, server, password_file, dump_filename):
        """Return a tuple of command template, command options, environment variables and standard input.

         Result is suitable for runner.sh().

        :type server: parallels.core.connections.database_servers.base.DatabaseServer
        :type password_file: str | unicode
        :type dump_filename: str | unicode
        :rtype: tuple
        """
        pass

    def create_password_file(self, server, database_name):
        """Create a file with database access credentials.

        :type server: parallels.core.connections.database_servers.base.DatabaseServer
        :type database_name: str | unicode
        :rtype: str | unicode
        """
        pass

    def handle_failed_restore_command(self, source_server, target_server, exception):
        """Handle failed restore command execution (if restore command returned non-zero exit code)

        :type source_server: parallels.core.connections.database_servers.base.DatabaseServer
        :type target_server: parallels.core.connections.database_servers.base.DatabaseServer
        :type exception: parallels.core.runners.exceptions.non_zero_exit_code.NonZeroExitCodeException
        :raise: Exception
        :rtype: None
        """
        raise exception

    def _get_dump_options(self, server, dump_filename):
        return dict(
            src_host=server.host(), src_port=server.port(),
            src_admin=server.user(), src_password=server.password(),
            db_name=self.source_database_name, dump_tmpname=dump_filename)

    def _get_restore_options(self, server, existing_options):
        options = existing_options.copy()
        connection_info = _get_db_connection_info(server, self.target_database_name)
        destination_credentials = _get_target_db_connection_settings(connection_info)
        options.update(destination_credentials)
        return options


class PostgresqlTransferConfiguration(DatabaseTransferConfiguration):
    """Configuration options and commands for transferring PostgreSQL databases."""

    def get_dump_command(self, server, password_file, dump_filename):
        """Return a tuple of command template, command options, environment variables and standard input.

         Result is suitable for runner.sh().

        :type server: parallels.core.connections.database_servers.base.DatabaseServer
        :type password_file: str | unicode
        :type dump_filename: str | unicode
        :rtype: tuple
        """
        command = ''
        env = {}
        stdin = None
        if password_file:
            env['PGPASSFILE'] = password_file
        else:
            env['PGUSER'] = server.user()
            env['PGDATABASE'] = self.source_database_name
            if server.password():
                command += u'PGPASSWORD="$(cat)" '
                stdin = server.password()

        command += self.dump_command_template
        if server.host() != 'localhost':
            command += u" -h {src_host} -p {src_port}"

        if self._is_ignore_version_option_available(server):
            command += u" --ignore-version"

        command += u" > {dump_tmpname}"
        options = self._get_dump_options(server, dump_filename)
        return command, options, env, stdin

    def get_restore_command(self, server, password_file, dump_filename):
        """Return a tuple of command template, command options, environment variables and standard input.

         Result is suitable for runner.sh().

        :type server: parallels.core.connections.database_server.DatabaseServer
        :type password_file: str | unicode
        :type dump_filename: str | unicode
        :rtype: tuple
        """
        env = {
            'PGPASSFILE': password_file
        }
        stdin = None
        restore_command = self.restore_command_template
        if server.host() != 'localhost':
            restore_command += " -h {dst_host} -p {dst_port}"
        command = u"%s < {dump_filename}" % (restore_command,)
        options = self._get_restore_options(server, dict(dump_filename=dump_filename))
        return command, options, env, stdin

    def create_password_file(self, server, database_name):
        """Create a file with database access credentials.

        :type server: parallels.core.connections.database_servers.base.DatabaseServer
        :type database_name: str | unicode
        :rtype: str | unicode
        """
        connection_info = _get_db_connection_info(server, database_name)
        if connection_info.password is not None:
            return _create_pgpass_file(server, connection_info)
        else:
            return None

    @property
    def dump_command_template(self):
        """Return the dump command with basic options independent from host or database name."""
        raise NotImplementedError()

    @property
    def restore_command_template(self):
        """Return the restore command with basic options independent from host or database name."""
        raise NotImplementedError()

    @staticmethod
    @cached
    def _is_ignore_version_option_available(server):
        """
        :type server: parallels.core.connections.database_servers.base.DatabaseServer
        """
        with server.runner() as runner:
            assert isinstance(runner, BaseRunner)
            help = runner.execute_command(
                'pg_dump --help', execution_options=ExecutionOptions(log_output=False)
            ).stdout
            return '--ignore-version' in help


class PostgresqlPlaintextTransferConfiguration(PostgresqlTransferConfiguration):
    """Commands and options for transferring a PostgreSQL database in plaintext format."""
    def __init__(self, source_database_name, target_database_name):
        super(PostgresqlTransferConfiguration, self).__init__(source_database_name, target_database_name)
        self._original_owner = None

    @property
    def dump_command_template(self):
        return u"pg_dump -U {src_admin} --no-acl --no-owner --clean {db_name}"

    @property
    def restore_command_template(self):
        return u"psql -U {dst_login} -d {dst_db_name}"

    def set_original_owner(self, owner_role):
        self._original_owner = owner_role

    def get_original_owner(self):
        return self._original_owner

    def get_restore_command(self, server, password_file, dump_filename):
        """Return a tuple of command template, command options, environment variables and standard input.

         Result is suitable for runner.sh().

        :type server: parallels.core.connections.database_servers.base.DatabaseServer
        :type password_file: str | unicode
        :type dump_filename: str | unicode
        :rtype: tuple
        """
        psql_command = self._get_psql(server)
        env = {
            "PGPASSFILE": password_file
        }
        stdin = None

        dump_tweak_command = u'''
            sed -i "1a \\
                    SET ROLE {original_owner};"'''
        command = u'%s {dump_filename}; %s < {dump_filename};' % (dump_tweak_command, psql_command)

        existing_options = dict(
            original_owner=self.get_original_owner(),
            dump_filename=dump_filename,
            dump_tweak_command=dump_tweak_command
        )

        options = self._get_restore_options(server, existing_options)
        return command, options, env, stdin

    def _get_psql(self, server):
        """Construct a basic psql command with access credentials."""
        restore_command = self.restore_command_template
        if server.host() != 'localhost':
            restore_command += " -h {dst_host} -p {dst_port}"
        return restore_command


class PostgresqlBinaryFormatTransferConfiguration(PostgresqlTransferConfiguration):
    """Commands and options for transferring a PostgreSQL database in binary format."""
    @property
    def dump_command_template(self):
        return u"pg_dump -U {src_admin} --format=custom --blobs --no-owner {db_name}"

    @property
    def restore_command_template(self):
        return u"pg_restore -v -U {dst_login} -d {dst_db_name} -c"


class MysqlTransferConfiguration(DatabaseTransferConfiguration):
    """Configuration, commands and options for transferring a MySQL database."""

    def get_dump_command(self, server, password_file, dump_filename):
        """Return a tuple of command template, command options, environment variables and standard input.

         The result is suitable for runner.sh().

        :type server: parallels.core.connections.database_servers.base.DatabaseServer
        :type password_file: str | unicode
        :type dump_filename: str | unicode
        :rtype: tuple
        """
        command = ''
        stdin = None
        if server.host == 'localhost' and server.port == 3306:
            # a workaround for Plesk
            command += u'MYSQL_PWD="$(cat /etc/psa/.psa.shadow)" '
        elif server.password():
            command += u'MYSQL_PWD="$(cat)" '
            stdin = server.password()
        command += (
            u"mysqldump --no-defaults -h {src_host} -P {src_port} -u{src_admin} --quick --quote-names "
            u"--add-drop-table ")

        if _mysql_supports_utf8mb4(server):
            command += u"--default-character-set=utf8mb4 --set-charset "
        else:
            command += u"--default-character-set=utf8 --set-charset "

        mysqldump_capabilities = _get_mysqldump_capabilities(server)
        if mysqldump_capabilities.has_routines:
            command += u"--routines "
        if mysqldump_capabilities.has_events:
            command += u"--events "

        command += u"{db_name} > {dump_tmpname}"
        options = self._get_dump_options(server, dump_filename)
        return command, options, {}, stdin

    def get_restore_command(self, server, password_file, dump_filename):
        """Return a tuple of command template, command options, environment variables and standard input.

         The result is suitable for runner.sh().

        :type server: parallels.core.connections.database_servers.base.DatabaseServer
        :type password_file: str | unicode
        :type dump_filename: str | unicode
        :rtype: tuple
        """
        command, stdin = _get_mysql_restore_command_template(
            server.host(), server.port(), server.password(), password_file
        )
        command += u' < {dump_filename}'
        options = self._get_restore_options(server, dict(dump_filename=dump_filename))
        options['option_file'] = password_file
        return command, options, {}, stdin

    def create_password_file(self, server, database_name):
        """Create a file with database access credentials.

        :type server: parallels.core.connections.database_servers.base.DatabaseServer
        :type database_name: str | unicode
        :rtype: str | unicode
        """
        password_file = _create_mysql_options_file(server, database_name)
        return password_file

    def handle_failed_restore_command(self, source_server, target_server, exception):
        """Handle failed restore command execution (if restore command returned non-zero exit code)

        :type source_server: parallels.core.connections.database_servers.base.DatabaseServer
        :type target_server: parallels.core.connections.database_servers.base.DatabaseServer
        :type exception: parallels.core.runners.exceptions.non_zero_exit_code.NonZeroExitCodeException
        :raise: Exception
        :rtype: None
        """
        # Handle known incompatibility between MySQL 5.5 and MySQL 5.1, when migrating from 5.5 to 5.1
        mysql_error_message_lower = exception.stderr.lower()
        if (
            "unknown character set:" in mysql_error_message_lower or
            "unknown collation:" in mysql_error_message_lower
        ):
            raise MigrationNoRepeatError(messages.MYSQL_TARGET_VERSION_DOES_NOT_HAVE_ENCODING.format(
                restoration_error=exception.stderr
            ))

        raise exception

    @staticmethod
    def _string_version_to_int_version(version):
        try:
            return [int(v) for v in version[:2]]
        except ValueError:
            return None


class LinuxDatabaseContentTransfer(object):
    """The object, that performs migration of database content."""
    def __init__(self, db_transfer_configuration, keyfile):
        self.keyfile = keyfile
        self.transfer_config = db_transfer_configuration

    def transfer(self, source_server, target_server, source_database_name, target_database_name):
        """Perform migration of database content.

        :type source_server: parallels.core.connections.database_servers.base.DatabaseServer
        :type target_server: parallels.core.connections.database_servers.base.DatabaseServer
        :type source_database_name: str | unicode
        :type target_database_name: str | unicode
        :rtype: None
        """
        dump_password_file = self.transfer_config.create_password_file(source_server, source_database_name)
        source_dump_path = self._create_dump(source_server, dump_password_file, source_database_name)
        target_dump_filename = self._transfer_dump_file(
            source_dump_path, source_server, target_server, target_database_name
        )
        restore_password_file = self.transfer_config.create_password_file(target_server, target_database_name)
        self._restore(target_dump_filename, source_server, target_server, restore_password_file)

        with source_server.runner() as runner:
            if dump_password_file is not None:
                runner.remove_file(dump_password_file)
            runner.remove_file(source_dump_path)

        with target_server.runner() as runner:
            runner.remove_file(restore_password_file)
            runner.remove_file(target_dump_filename)

    def _create_dump(self, server, password_file, database_name):
        """Dump source database to disk; return dump filename."""
        with server.runner() as runner:
            assert isinstance(runner, BaseRunner)
            runner.mkdir(server.get_session_file_path('db-dumps'))
            dump_filename = server.get_session_file_path(posixpath.join('db-dumps', '%s.sql' % database_name))
            runner.remove_file(dump_filename)
            runner.execute_command('touch {filename}', dict(filename=dump_filename))
            runner.execute_command('chmod 600 {filename}', dict(filename=dump_filename))
            command, options, env, stdin = self.transfer_config.get_dump_command(server, password_file, dump_filename)
            runner.execute_command(command, options, ExecutionOptions(stdin_content=stdin, environment_variables=env))
        return dump_filename

    def _transfer_dump_file(self, source_dump_path, source_server, target_server, database_name):
        """Transfer dump file from source to target server."""

        with target_server.runner() as runner:
            assert isinstance(runner, BaseRunner)
            runner.mkdir(target_server.get_session_file_path('db-dumps'))
            target_dump_tmpname = target_server.get_session_file_path(
                posixpath.join('db-dumps', '%s.sql' % database_name)
            )
            runner.remove_file(target_dump_tmpname)
            runner.execute_command(
                u"scp -i {keyfile} -P {port}"
                u" -o PasswordAuthentication=no -o StrictHostKeyChecking=no -o GSSAPIAuthentication=no"
                u" {src_server_user}@{src_server_ip}:{source_dump_tmpname} {target_dump_tmpname}",
                dict(
                    keyfile=self.keyfile,
                    src_server_user=source_server.utilities_server.user(),
                    src_server_ip=source_server.utilities_server.ip(),
                    source_dump_tmpname=source_dump_path,
                    target_dump_tmpname=target_dump_tmpname,
                    port=source_server.utilities_server.settings().ssh_auth.port
                )
            )
        return target_dump_tmpname

    def _restore(self, dump_filename, source_server, target_server, password_file):
        command, options, env, stdin = self.transfer_config.get_restore_command(
            target_server, password_file, dump_filename
        )
        with target_server.runner() as runner:
            assert isinstance(runner, BaseRunner)
            try:
                runner.execute_command(
                    command, options, ExecutionOptions(stdin_content=stdin, environment_variables=env)
                )
            except NonZeroExitCodeException as e:
                self.transfer_config.handle_failed_restore_command(source_server, target_server, e)


def restore_db_from_dump_linux(
    server, database_name, dump_filename, login=None, password=None, controlled_interruption=None
):
    """
    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :type database_name: str | unicode
    :type dump_filename: str | unicode
    :type login: str | unicode | None
    :type password: str | unicode | None
    :type controlled_interruption: parallels.core.runners.entities.ControlledInterruption | None
    """
    connection_info = _get_db_connection_info(server, database_name)
    command_parameters = dict(dump_filename=dump_filename)
    destination_credentials = _get_target_db_connection_settings(connection_info, login, password)
    command_parameters.update(destination_credentials)
    env = {}
    stdin = None
    if server.type() == DatabaseServerType.MYSQL:
        passwd_file = _create_mysql_options_file(server, database_name, login, password)
        restore_command, stdin = _get_mysql_restore_command_template(
            server.host(), server.port(), server.password(), passwd_file
        )
        command_parameters['option_file'] = passwd_file
    elif server.type() == DatabaseServerType.POSTGRESQL:
        passwd_file = _create_pgpass_file(server, connection_info)
        restore_command, restore_env = _get_binary_pg_restore_command_template(
            server.host(), passwd_file
        )
        env.update(restore_env)
    else:
        raise Exception(messages.CANNOT_TRANSFER_DATABASE_OF_UNSUPPORTED_TYPE % server.type())

    with server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        runner.execute_command(
            restore_command + u" < {dump_filename}", command_parameters,
            ExecutionOptions(
                stdin_content=stdin, environment_variables=env,
                controlled_interruption=controlled_interruption
            )
        )
        runner.remove_file(passwd_file)


def restore_mysql_dump_under_user_linux(
    server, system_user, temp_directory,
    database_name, database_login, database_password,
    dump_file,
    controlled_interruption=None
):
    """Restore MySQL dump running MySQL client under specified system user in a secure way.
    
    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :type system_user: str | unicode 
    :type temp_directory: str | unicode 
    :type database_name: str | unicode 
    :type database_login: str | unicode 
    :type database_password: str | unicode 
    :type dump_file: str | unicode 
    :type controlled_interruption: parallels.core.runners.entities.ControlledInterruption | None
    :rtype: None 
    """
    # 1. Create options file in safe location, available to root/pssadm only.
    # Assume database dump is in the same safe location.
    options_file = _create_mysql_options_file(server, database_name, database_login, database_password)

    # 2. Generate unique names for options and dump files in temporary directory
    # from which system user could read the file (e.g. virtual host of subscription).
    user_options_file = temp_filename(temp_directory, 'migration-mysql-options')
    user_dump_file = temp_filename(temp_directory, 'migration-mysql-dump')

    # 3. Move options file (with sensitive data) so it is readable only by system user of subscription.
    with server.runner() as runner:
        move_file_secure(runner, system_user, source_file_path=options_file, target_file_path=user_options_file)

    try:
        # 4. Move database dump file to a directory, from which the system user could read the file
        # (e.g. virtual host of subscription), and make the file readable only by the system user.
        with server.runner() as runner:
            move_file_secure(runner, system_user, dump_file, user_dump_file)

        try:
            mysql_restore_command = unix_utils.format_command(
                u"mysql --defaults-file={options_file} -h {host} -P {port} -u{login} {database} < {dump}",
                host=server.host(), port=server.port(), login=database_login, database=database_name,
                dump=user_dump_file, options_file=user_options_file
            )
            with server.runner() as runner:
                assert isinstance(runner, BaseRunner)
                # 5. Run database restoration with "mysql" utility under system user.
                runner.execute_command(
                    u'su {username} --shell /bin/sh -c {command}',
                    dict(
                        username=system_user,
                        command=mysql_restore_command
                    ),
                    execution_options=ExecutionOptions(controlled_interruption=controlled_interruption)
                )
        finally:
            try:
                os.unlink(user_dump_file)
            except Exception as e:
                logger.exception()
                logger.fwarn(messages.FAILED_TO_REMOVE_DUMP_FILE, filename=user_dump_file, reason=e)
    finally:
        try:
            os.unlink(user_options_file)
        except Exception as e:
            logger.exception()
            logger.fwarn(messages.FAILED_TO_REMOVE_OPTIONS_FILE, filename=user_options_file, reason=e)


def restore_mysql_dump_under_user_windows(
    server, system_user, temp_directory,
    database_name, database_login, database_password,
    dump_file,
    controlled_interruption=None
):
    """Restore MySQL dump running MySQL client under specified system user in a secure way.

    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :type system_user: str | unicode 
    :type temp_directory: str | unicode 
    :type database_name: str | unicode 
    :type database_login: str | unicode 
    :type database_password: str | unicode 
    :type dump_file: str | unicode 
    :type controlled_interruption: parallels.core.runners.entities.ControlledInterruption | None
    :rtype: None 
    """
    with server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        runner.execute_command("icacls {dump} /grant {user}:F", dict(dump=dump_file, user=system_user))

    user_dump_file = temp_filename(temp_directory, 'migration-mysql-dump')
    user_options_file = temp_filename(temp_directory, 'migration-mysql-options')

    os.rename(dump_file, user_dump_file)

    try:
        options_file = _create_mysql_options_file(server, database_name, database_login, database_password)

        with server.runner() as runner:
            assert isinstance(runner, BaseRunner)
            runner.execute_command("icacls {dump} /grant {user}:F", dict(dump=options_file, user=system_user))

        os.rename(options_file, user_options_file)

        try:
            # Call wrapper around the 'mysql' migrator, which executes the 'mysql'
            # with dump restore options under specified system user
            config = dict(
                host=server.host(),
                port=server.port(),
                defaults_file=user_options_file,
                database=database_name,
                dump=user_dump_file
            )

            config_json = json.dumps(config, indent=4)

            with server.runner() as runner:
                assert isinstance(runner, BaseRunner)
                extension_id = os.getenv('PLESK_EXTENSION_ID')
                if is_empty(extension_id):
                    extension_id = 'panel-migrator'
                runner.execute_command(
                    'plesk bin extension --exec {extension_id} restore-mysql-dump.php {user}',
                    dict(extension_id=extension_id, user=system_user),
                    ExecutionOptions(
                        stdin_content=config_json,
                        controlled_interruption=controlled_interruption
                    )
                )
        finally:
            try:
                os.unlink(user_options_file)
            except Exception as e:
                logger.exception()
                logger.fwarn(messages.FAILED_TO_REMOVE_OPTIONS_FILE, filename=user_options_file, reason=e)
    finally:
        try:
            os.unlink(user_dump_file)
        except Exception as e:
            logger.exception()
            logger.fwarn(messages.FAILED_TO_REMOVE_DUMP_FILE, filename=user_dump_file, reason=e)


def _get_target_db_connection_settings(connection_settings, login=None, password=None):
    """
    :type connection_settings: parallels.core.utils.database_utils.DatabaseConnectionInfo
    :type login: str | unicode | None
    :type password: str | unicode | None
    :rtype: dict[str | unicode, str | unicode]
    """
    d = connection_settings.as_dictionary()

    if login is not None:
        d['login'] = login
    if password is not None:
        d['password'] = password

    return add_dict_key_prefix('dst_', d)


def _get_db_connection_info(server, db_name):
    """Return an object that defines database connection.

    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :type db_name: str | unicode
    :rtype: parallels.core.utils.database_utils.DatabaseConnectionInfo
    """
    return DatabaseConnectionInfo(
        host=server.host(), port=server.port(), login=server.user(),
        password=server.password(), db_name=db_name
    )


def _get_mysql_restore_command_template(host, port, password=None, option_file=None):
    """
    :type host: str | unicode
    :type port: str | unicode
    :type password: str | unicode
    :type option_file: str | unicode
    :rtype: tuple
    """
    restore_command = (
        u"%s mysql %s -h {dst_host} -P {dst_port} -u{dst_login} {dst_db_name}")
    env = options = ''
    stdin = None
    if option_file:
        env = ''
        options = '--defaults-file={option_file}'
    elif host == 'localhost' and port == 3306:
        env = u'MYSQL_PWD="$(cat /etc/psa/.psa.shadow)"'
        options = u'--no-defaults'
    elif password:
        env = u'MYSQL_PWD="$(cat)"'
        options = u'--no-defaults'
        stdin = password
    return restore_command % (env, options), stdin


def _create_mysql_options_file(server, file_id, login=None, password=None):
    """Create a mysql options file and upload it to the remote server."""
    filename = 'my_{host}_{file_id}.cnf'.format(host=server.host(), file_id=file_id)
    connection_str = u"[client]\nuser={login}\npassword='{password}'".format(
        login=login if login is not None else server.user(),
        password=password if password is not None else server.password()
    )
    remote_filename = server.get_session_file_path(filename)
    with server.runner() as runner:
        runner.upload_file_content(remote_filename, connection_str)
    return remote_filename


def _get_binary_pg_restore_command_template(hostname, passwd_file):
    """
    :type hostname: str | unicode
    :type passwd_file: str | unicode | None
    :rtype: str | unicode
    """
    env = {'PGPASSFILE': passwd_file}
    restore_command = u"pg_restore -v -U {dst_login} -d {dst_db_name} -c"
    if hostname != 'localhost':
        restore_command += " -h {dst_host} -p {dst_port}"
    return restore_command, env


def _create_pgpass_file(server, connection_info):
    """Create a PostgreSQL password file and upload it to the remote server.

    :type connection_info: parallels.core.utils.database_utils.DatabaseConnectionInfo
    :rtype: str | unicode
    """
    filename = 'pgpass_{host}_{db_name}'.format(**connection_info.as_dictionary())
    connection_str = '*:*:{db_name}:{login}:{password}'.format(**connection_info.as_dictionary())

    remote_filename = server.get_session_file_path(filename)
    with server.runner() as runner:
        runner.upload_file_content(remote_filename, connection_str)
        # PostgreSQL requires password file not to have group permissions,
        # but by default migrator sets them so all files are accessible by both
        # backend and frontend. Revoke group permissions here for the file in explicit way -
        # it will never be used by frontend.
        runner.chmod(remote_filename, 0o600)

    return remote_filename


def restore_db_from_dump_windows(db_server, database_name, dump_filename):
    """Restore database dump file on Windows database server

    :type db_server: parallels.core.connections.database_servers.base.DatabaseServer
    :type database_name: str | unicode
    :type dump_filename: str | unicode
    """
    if db_server.type() == DatabaseServerType.MYSQL:
        restore_windows_mysql_dump(db_server, database_name, dump_filename)
    elif db_server.type() == DatabaseServerType.MSSQL:
        restore_windows_mssql_dump(db_server, database_name, dump_filename)
    else:
        logger.error(messages.DATABASE_UNSUPPORTED_TYPE_AND_HENCE_NOT)


def copy_db_content_windows(database_info, mssql_copy_method=MSSQLCopyMethod.AUTO):
    """Copy database contents from one Windows database server to another

    :type database_info: parallels.core.utils.database_utils.DatabaseInfo
    :type mssql_copy_method: str | unicode
    """
    source = database_info.source_database_server
    target = database_info.target_database_server
    logger.info(messages.LOG_COPY_DB_CONTENT.format(
        db_name=database_info.source_database_name,
        source=source.description(),
        target=target.description()
    ))

    if source.type() == DatabaseServerType.MYSQL:
        _copy_db_content_windows_mysql(database_info)
    elif source.type() == DatabaseServerType.MSSQL:
        _copy_db_content_mssql(database_info, mssql_copy_method)
    else:
        logger.error(messages.DATABASE_UNSUPPORTED_TYPE_AND_HENCE_NOT)


# MSSQL server fails to restore multiple databases at the same time with errors like:
# "Cannot open backup device '...'. Operating system error 32(The process cannot access the file
# because it is being used by another process.)", so we have "synchronized" here.
@synchronized
def _copy_db_content_mssql(database_info, mssql_copy_method):
    """Copy MSSQL database contents from one Windows database server to another

    :type database_info: parallels.core.utils.database_utils.DatabaseInfo
    :type mssql_copy_method: str | unicode
    """
    source = database_info.source_database_server
    target = database_info.target_database_server

    mssql_copy_method = get_mssql_copy_method(source, target, mssql_copy_method)
    if mssql_copy_method == MSSQLCopyMethod.TEXT:
        # Copy databases with Plesk dbbackup.exe utility, '--copy' command, which uses text dump created with SMO
        _copy_db_content_windows_mssql_text_backup(database_info)
    elif mssql_copy_method == MSSQLCopyMethod.NATIVE:
        # Copy databases with MSSQL native backups - T-SQL "BACKUP" and "RESTORE" procedures
        _copy_db_content_windows_mssql_native_backup(database_info)
    else:
        assert False


def get_mssql_copy_method(source_database_server, target_database_server, mssql_copy_method=MSSQLCopyMethod.AUTO):
    """Get MSSQL copy method to be used for specified MSSQL database server

    Considers the following factors:
    1. Global configuration option in config.ini file.
    2. If we have physical access to both source and target servers.

    :type mssql_copy_method: str | unicode
    :type source_database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :type target_database_server: parallels.core.connections.database_servers.base.DatabaseServer
    """
    if mssql_copy_method == MSSQLCopyMethod.AUTO:
        # Automatically select available copy method. If we have access to both source and target servers - use
        # native backups, otherwise use SMO dump that does not require direct access
        if source_database_server.physical_server is not None and target_database_server.physical_server is not None:
            return MSSQLCopyMethod.NATIVE
        else:
            return MSSQLCopyMethod.TEXT
    else:
        return mssql_copy_method


def _copy_db_content_windows_mysql(database_info):
    """Copy MySQL database contents from one Windows database server to another

    :type database_info: parallels.core.utils.database_utils.DatabaseInfo
    """
    source = database_info.source_database_server
    target = database_info.target_database_server

    with target.runner() as runner_target, source.runner() as runner_source:
        assert isinstance(runner_source, BaseRunner)
        dump_filename = 'db_backup_%s_%s.sql' % (
            database_info.subscription_name.encode('idna'), database_info.source_database_name
        )
        source_dump_filename = source.panel_server.get_session_file_path(dump_filename)
        target_dump_filename = target.panel_server.get_session_file_path(dump_filename)

        if not is_windows_mysql_client_configured(target.panel_server):
            raise MigrationError(messages.MYSQL_CLIENT_BINARY_WAS_NOT_FOUND % (
                target.description(), database_info.source_database_name
            ))

        mysqldump_capabilities = _get_mysqldump_capabilities(source)
        runner_source.execute_command(
            u'{path_to_mysqldump} -h {host} -P {port} -u{user} -p{password} --default-character-set=' + (
                u'utf8mb4' if _mysql_supports_utf8mb4(source) else u'utf8'
            ) + u' --set-charset {database_name} --result-file={source_dump_filename}' + (
                u' --skip-secure-auth' if source.panel_server.mysql_use_skip_secure_auth() else u''
            ) + (
                u" --routines" if mysqldump_capabilities.has_routines else u''
            ) + (
                u" --events" if mysqldump_capabilities.has_events else u''
            ),
            dict(
                path_to_mysqldump=source.panel_server.get_path_to_mysqldump(),
                host=source.host(),
                port=source.port(),
                user=source.user(),
                password=source.password(),
                database_name=database_info.source_database_name,
                source_dump_filename=source_dump_filename
            )
        )

        logger.debug(messages.COPY_DATABASE_DUMP_FROM_SOURCE_TARGET)
        transfer_file(runner_source, source_dump_filename, runner_target, target_dump_filename)

        restore_windows_mysql_dump(target, database_info.target_database_name, target_dump_filename)

        logger.debug(messages.LOG_REMOVE_DB_DUMP_FILES % (
            source_dump_filename, target_dump_filename
        ))
        runner_source.remove_file(source_dump_filename)
        runner_target.remove_file(target_dump_filename)


def restore_windows_mysql_dump(
    db_server, database_name, dump_filename, mysql_client=None, login=None, password=None,
    controlled_interruption=None
):
    """Restore MySQL database dump on specified database server

    :type db_server: parallels.core.connections.database_servers.base.DatabaseServer
    :type database_name: str | unicode
    :type dump_filename: str | unicode
    :type login: str | unicode | None
    :type password: str | unicode | None
    :type controlled_interruption: parallels.core.runners.entities.ControlledInterruption | None
    """
    logger.debug(messages.RESTORE_DATABASE_DUMP_TARGET_SERVER)

    if mysql_client is None:
        mysql_client = get_windows_mysql_client(db_server.panel_server)

    with db_server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        runner.execute_command(
            u'{mysql_client} --no-defaults -h {host} -P {port} -u{login} -p{password} --default-character-set=utf8 {database_name} '
            u'-e {query}',
            dict(
                mysql_client=mysql_client,
                host=db_server.host(),
                port=db_server.port(),
                login=login if login is not None else db_server.user(),
                password=password if password is not None else db_server.password(),
                database_name=database_name,
                query='source %s' % dump_filename
            ),
            ExecutionOptions(
                controlled_interruption=controlled_interruption
            )
        )


def _copy_db_content_windows_mssql_text_backup(database_info):
    """Copy MSSQL database contents from one Windows database server to another, using SMO text dump

    :type database_info: parallels.core.utils.database_utils.DatabaseInfo
    """
    source = database_info.source_database_server
    target = database_info.target_database_server
    with target.panel_server.runner() as runner_target:
        assert isinstance(runner_target, BaseRunner)
        runner_target.execute_command(
            u'{dbbackup_path} --copy -copy-if-logins-exist -with-data -src-server={src_host} '
            u'-server-type={db_type} -src-server-login={src_admin} -src-server-pwd={src_password} '
            u'-src-database={src_database_name} -dst-server={dst_host} -dst-server-login={dst_admin} '
            u'-dst-server-pwd={dst_password} -dst-database={dst_database_name}',
            dict(
                dbbackup_path=_get_dbbackup_bin(target.panel_server),
                db_type=source.type(),
                src_host=get_mssql_host_for_target_server(source),
                src_admin=source.user(),
                src_password=source.password(),
                dst_host=target.host(),
                dst_admin=target.user(),
                dst_password=target.password(),
                src_database_name=database_info.source_database_name,
                dst_database_name=database_info.target_database_name,
            )
        )


def _copy_db_content_windows_mssql_native_backup(database_info):
    """Copy MSSQL database contents from one Windows database server to another, using native MSSQL dump

    :type database_info: parallels.core.utils.database_utils.DatabaseInfo
    """
    source = database_info.source_database_server
    target = database_info.target_database_server
    dump_filename = 'db_backup_%s_%s.sql' % (
        database_info.subscription_name.encode('idna'), database_info.source_database_name
    )
    dumps_dir = 'mssql-dumps'
    source_dumps_dir = _create_dumps_dir(source, dumps_dir)
    if target.is_local():
        with target.runner() as runner:
            target_dumps_dir = get_windows_temp_dump_dir(runner)
    else:
        target_dumps_dir = _create_dumps_dir(target, dumps_dir)
    source_dump_filename = ntpath.join(source_dumps_dir, dump_filename)
    target_dump_filename = ntpath.join(target_dumps_dir, dump_filename)

    if source.physical_server is None:
        raise MigrationError(
            messages.MSSQL_DATABASE_NATIVE_DUMP_NO_PHYSICAL_ACCESS.format(server=source.description())
        )

    if target.physical_server is None:
        raise MigrationError(
            messages.MSSQL_DATABASE_NATIVE_DUMP_NO_PHYSICAL_ACCESS.format(server=target.description())
        )

    with source.panel_server.runner() as runner_panel_source:
        # Execute SQL command connecting from panel server, as panel server must have access to
        # SQL server by specified connection data and credentials. It is a bit less reliable to
        # execute SQL command directly from MSSQL server: it could be impossible to resolve hostname of MSSQL
        # server from itself (while it must be resolved from panel server), there could be firewall rules
        # blocking connections to itself (while panel server must always have firewall exceptions to control
        # MSSQL server), and so on.
        connection_settings = MSSQLConnectionSettings(
            host=source.host(), user=source.user(), password=source.password(),
        )
        runner_panel_source.remove_file(source_dump_filename)
        runner_panel_source.mssql_execute(
            connection_settings,
            "BACKUP DATABASE [%s] TO DISK=%%(file)s" % database_info.source_database_name,
            dict(file=source_dump_filename)
        )

    with source.physical_server.runner() as runner_source, target.runner() as runner_target:
        transfer_file(runner_source, source_dump_filename, runner_target, target_dump_filename)

    restore_windows_mssql_dump(target, database_info.target_database_name, target_dump_filename)

    with source.physical_server.runner() as runner:
        runner.remove_file(source_dump_filename)

    with target.physical_server.runner() as runner:
        runner.remove_file(target_dump_filename)


def restore_windows_mssql_dump(db_server, database_name, dump_filename):
    """Restore MSSQL native database dump on specified database server

    :type db_server: parallels.core.connections.database_servers.base.DatabaseServer
    :type database_name: str | unicode
    :type dump_filename: str | unicode
    """
    if db_server.physical_server is None:
        raise MigrationError(
            messages.MSSQL_DATABASE_NATIVE_DUMP_NO_PHYSICAL_ACCESS.format(server=db_server.description())
        )
    with db_server.panel_server.runner() as runner_target:
        assert isinstance(runner_target, BaseRunner)
        runner_target.execute_command(
            r'{dbbackup_path} --restore -server={dst_host} '
            r'-server-type=mssql -server-login={dst_admin} -server-pwd={dst_password} '
            r'-database={database_name} -backup-path={backup_path}',
            dict(
                dbbackup_path=_get_dbbackup_bin(db_server.panel_server),
                dst_host=db_server.host(),
                dst_admin=db_server.user(),
                dst_password=db_server.password(),
                database_name=database_name,
                backup_path=dump_filename
            )
        )


@cached
def _create_dumps_dir(server, dumps_dir):
    """
    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: str
    """
    full_dumps_dir = server.get_session_file_path(dumps_dir)
    with server.runner() as runner:
        runner.mkdir(full_dumps_dir)

        try:
            set_mssql_dumps_dir_permissions(server, full_dumps_dir)
        except Exception as e:
            logger.exception()
            logger.fwarn(
                messages.FAILED_TO_SET_DBDUMP_DIR_PERMISSIONS,
                reason=e, server=server.description(), directory=full_dumps_dir
            )
        finally:
            # In any case, return path to created dumps directory, even if there were some
            # issues when setting permissions
            return full_dumps_dir


def set_mssql_dumps_dir_permissions(server, dumps_dir):
    """Set permissions for MSSQL dumps directory, so MSSQL server can read/write to it

    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :type dumps_dir: str | unicode
    :rtype: None
    """
    with server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        try:
            if (
                '/inheritance' in runner.execute_command(
                    'icacls /?',
                    execution_options=ExecutionOptions(
                        output_encoding=runner.native_output_encoding,
                        log_output=False
                    )
                ).stdout
            ):
                # We can use "icacls" utility for configuring permissions
                # For old Windows versions (for example, 2003) this code will be simply skipped
                # and directory permissions will inherit session directory one's. In that case
                # it is a customer's responsibility to configure secure session directory.
                _, instance_name, _ = split_mssql_host_parts_str(server.host())
                if instance_name:
                    mssql_instance_str = '$%s' % instance_name
                else:
                    mssql_instance_str = ''

                user = get_from_registry(
                    runner, [
                        r'HKLM\SYSTEM\CurrentControlSet\services\MSSQL%s' % mssql_instance_str,
                        r'HKLM\SYSTEM\CurrentControlSet\services\MSSQLSERVER%s' % mssql_instance_str
                    ], 'ObjectName'
                )

                if is_empty(user):
                    # No user defined - leave default permissions
                    return
                elif user == 'LocalSystem':
                    # Don't need to set permissions for LocalSystem in explicit way
                    return
                else:
                    if '\\' in user:
                        # By default, custom user for MSSQL server has ".\" prefix in its name in registry.
                        # Get rid of that prefix, so ".\mycustomuser" becomes "mycustomuser"
                        user = user[user.find('\\') + 1:]

                        if is_empty(user):
                            return

                    runner.execute_command(
                        "icacls {dir} /inheritance:r", dict(dir=dumps_dir),
                        ExecutionOptions(output_encoding=runner.native_output_encoding)
                    )
                    permission_descriptions = [
                        # "S-1-5-32-544" is a security identifier of build-in Administrators group.
                        # Check https://support.microsoft.com/en-us/kb/243330 for the list of well-known
                        # security identifiers.
                        # We use SID instead of name, because the group may be named in other way, or renamed.
                        # For example, in German Windows it is called "Administratoren".
                        '*S-1-5-32-544:(OI)(CI)F',
                        'SYSTEM:(OI)(CI)F',
                        '%s:(OI)(CI)F' % user
                    ]
                    # Grant permissions to Administrators group, System, and MSSQL service user
                    for descr in permission_descriptions:
                        runner.execute_command(
                            "icacls {dir} /grant {descr}",
                            dict(dir=dumps_dir, descr=descr),
                            ExecutionOptions(output_encoding=runner.native_output_encoding)
                        )
        except Exception as e:
            logger.exception()
            logger.fwarn(
                messages.FAILED_TO_SET_DBDUMP_DIR_PERMISSIONS,
                reason=e, server=server.description(), directory=dumps_dir
            )


@cached
def _get_dbbackup_bin(server):
    with server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        dbbackup_bin = windows_utils.path_join(server.plesk_dir, r'admin\bin\dbbackup.exe')
        if not runner.file_exists(dbbackup_bin):
            dbbackup_bin = windows_utils.path_join(server.plesk_dir, r'admin\bin64\dbbackup.exe')
        return dbbackup_bin


def get_windows_mysql_client(target_server):
    return windows_utils.path_join(target_server.plesk_dir, 'MySQL\\bin\\mysql')


def is_windows_mysql_client_configured(target_server):
    mysql = get_windows_mysql_client(target_server)
    with target_server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        command_results = runner.execute_command(
            u'{mysql_bin} --version', dict(mysql_bin=mysql),
            ExecutionOptions(ignore_exit_code=True)
        )
        return command_results.exit_code == 0


def check_connection(database_server):
    """Check connection to specified database server

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :raises parallels.core.utils.database_utils.DatabaseServerConnectionException:
    """
    if is_empty(database_server.user()) or is_empty(database_server.password()):
        raise DatabaseServerConnectionException(
            messages.SERVER_IS_NOT_PROPERLY_CONFIGURED_TARGET.format(
                server=database_server.description()
            )
        )

    if database_server.type() == DatabaseServerType.MYSQL:
        check_mysql_connection(database_server)
    elif database_server.type() == DatabaseServerType.POSTGRESQL:
        check_postgresql_connection(database_server)
    elif database_server.type() == DatabaseServerType.MSSQL:
        check_mssql_connection(database_server)
    else:
        return


def check_mysql_connection(database_server):
    """Check connection to MySQL database server

    :type database_server: parallels.core.connections.database_server.DatabaseServer:
    :raises parallels.core.utils.database_utils.DatabaseServerConnectionException:
    """

    with database_server.runner() as runner:
        stdin = None
        if isinstance(runner, WindowsRunner):
            command = '{mysql} --silent --skip-column-names -h {host} -P {port} -u {user} -p{password} -e {query}'
        else:
            command = 'MYSQL_PWD="$(cat)" {mysql} --silent --skip-column-names -h {host} -P {port} -u {user} -e {query}'
            stdin = database_server.password()
        args = dict(
            mysql=database_server.mysql_bin,
            user=database_server.user(), password=database_server.password(),
            host=database_server.host(), port=database_server.port(),
            query='SELECT 1'
        )
        execution_results = runner.execute_command(
            command, args, ExecutionOptions(stdin_content=stdin, ignore_exit_code=True)
        )

        if execution_results.exit_code != 0 or execution_results.stdout.strip() != '1':
            raise DatabaseServerConnectionException(
                messages.DB_CONNECTION_FAILURE.format(
                    server=database_server.description(),
                    command=command.format(**args),
                    stdout=execution_results.stdout,
                    stderr=execution_results.stderr,
                    exit_code=execution_results.exit_code
                )
            )


def check_postgresql_connection(database_server):
    """Check connection to PostgreSQL database server

    :type database_server: parallels.core.connections.database_server.DatabaseServer:
    :raises parallels.core.utils.database_utils.DatabaseServerConnectionException:
    """
    command = 'PGUSER={user} PGPASSWORD="$(cat)" psql'
    if database_server.host() != 'localhost':
        command += " -h {host} -p {port}"
    command += " -dtemplate1 -A -t -q -c {query}"
    args = dict(
        user=database_server.user(), host=database_server.host(), port=database_server.port(),
        query='SELECT 1'
    )
    stdin = database_server.password()
    with database_server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        execution_results = runner.execute_command(
            command, args, ExecutionOptions(stdin_content=stdin, ignore_exit_code=True)
        )

    if execution_results.exit_code != 0 or execution_results.stdout.strip() != '1':
        raise DatabaseServerConnectionException(
            messages.DB_CONNECTION_FAILURE.format(
                server=database_server.description(),
                command=command.format(**args),
                stdout=execution_results.stdout,
                stderr=execution_results.stderr,
                exit_code=execution_results.exit_code
            )
        )


def check_mssql_connection(database_server, utilities_server=None):
    """Check connection to MSSQL database server

    :type database_server: parallels.core.connections.database_server.DatabaseServer:
    :type utilities_server: parallels.core.connections.server.Server
    :raises parallels.core.utils.database_utils.DatabaseServerConnectionException:
    """
    if utilities_server is None:
        utilities_server = database_server.utilities_server
    with utilities_server.runner() as runner:
        try:
            connection_settings = MSSQLConnectionSettings(
                host=database_server.host(),
                user=database_server.user(),
                password=database_server.password()
            )
            query = "SELECT 1 AS result"
            result = runner.mssql_query(connection_settings, query)
            if result != [{'result': 1}]:
                raise DatabaseServerConnectionException(safe_format(
                    messages.MSSQL_DB_CONNECTION_CHECK_RESULT,
                    server=database_server.description(),
                    query=query, result=repr(result)
                ))
        except DatabaseServerConnectionException:
            raise
        except Exception as e:
            logger.exception()
            raise DatabaseServerConnectionException(safe_format(
                messages.MSSQL_DB_CONNECTION_FAILURE,
                server=database_server.description(), reason=e
            ))


class DatabaseServerConnectionException(Exception):
    pass


def get_server_version(database_server):
    """Get version of specified database server. Now works for MySQL only.

    Returns tuple of version parts, for example, ('5', '5', '41-MariaDB'). If not able to determine the version,
    returns None.

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: tuple[str | unicode] | None
    """
    if database_server.type() == DatabaseServerType.MYSQL:
        with database_server.runner() as runner:
            assert isinstance(runner, BaseRunner)
            stdin = None
            if isinstance(runner, WindowsRunner):
                command = '{mysql} --silent --skip-column-names -h {host} -P {port} -u {user} -p{password} -e {query}'
            else:
                command = (
                    'MYSQL_PWD="$(cat)" {mysql} --silent --skip-column-names -h {host} -P {port} -u {user} -e {query}'
                )
                stdin = database_server.password()
            version_string = runner.execute_command(
                command,
                dict(
                    mysql=database_server.mysql_bin,
                    user=database_server.user(), password=database_server.password(),
                    host=database_server.host(), port=database_server.port(),
                    query='SELECT VERSION()'
                ),
                ExecutionOptions(stdin_content=stdin)
            ).stdout.strip()

            return version_string.split('.')
    else:
        return None


def list_databases(database_server):
    """List databases on specified server.

    Returns list of databases names or None if that function is not supported for that database server type.

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: set[str | unicode] | None
    """
    if database_server.type() == DatabaseServerType.MYSQL:
        return list_mysql_databases(database_server)
    elif database_server.type() == DatabaseServerType.POSTGRESQL:
        return list_postgresql_databases(database_server)
    elif database_server.type() == DatabaseServerType.MSSQL:
        return list_mssql_databases(database_server)
    else:
        return None


def list_mysql_databases(database_server):
    """List databases on specified MySQL server.

    Returns list of databases names.

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: set[str | unicode]
    """
    with database_server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        stdin = None
        if isinstance(runner, WindowsRunner):
            command = '{mysql} --silent --skip-column-names -h {host} -P {port} -u {user} -p{password} -e {query}'
        else:
            command = 'MYSQL_PWD="$(cat)" {mysql} --silent --skip-column-names -h {host} -P {port} -u {user} -e {query}'
            stdin = database_server.password()
        stdout = runner.execute_command(
            command,
            dict(
                mysql=database_server.mysql_bin,
                user=database_server.user(), password=database_server.password(),
                host=database_server.host(), port=database_server.port(),
                query='SHOW DATABASES'
            ),
            ExecutionOptions(stdin_content=stdin),
        ).stdout

        return set(split_nonempty_strs(stdout))


def list_postgresql_databases(database_server):
    """List databases on specified PostgreSQL server.

    Returns list of databases names or None if that function is not supported for that database server type.

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: set[str | unicode]
    """
    with database_server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        command = 'PGUSER={user} PGPASSWORD="$(cat)" psql'
        if database_server.host() != 'localhost':
            command += " -h {host} -p {port}"
        command += " -dtemplate1 -A -t -q -c {query}"
        stdin = database_server.password()
        stdout = runner.execute_command(
            command,
            dict(
                user=database_server.user(),
                host=database_server.host(), port=database_server.port(),
                query='SELECT datname FROM pg_database'
            ),
            ExecutionOptions(stdin_content=stdin)
        ).stdout

        return set(split_nonempty_strs(stdout))


def list_mssql_databases(database_server):
    """List databases on specified MSSQL server.

    Returns list of databases names or None if that function is not supported for that database server type.

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: set[str | unicode]
    """
    with database_server.panel_server.runner() as runner:
        connection_settings = MSSQLConnectionSettings(
            host=database_server.host(),
            user=database_server.user(),
            password=database_server.password()
        )
        result = runner.mssql_query(connection_settings, "SELECT [name] FROM master.dbo.sysdatabases")
        return {row['name'] for row in result}


def list_database_users(database_server):
    """List databases users on specified server.

    Returns list of database user names or None if that function is not supported for that database server type.

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: set[str | unicode] | None
    """
    if database_server.type() == DatabaseServerType.MYSQL:
        return list_mysql_database_users(database_server)
    else:
        return None


def list_mysql_database_users(database_server):
    """List database users on specified MySQL server.

    Returns list of databases names.

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: set[str | unicode]
    """
    with database_server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        stdin = None
        if isinstance(runner, WindowsRunner):
            command = '{mysql} --silent --skip-column-names -h {host} -P {port} -u {user} -p{password} -e {query}'
        else:
            command = 'MYSQL_PWD="$(cat)" {mysql} --silent --skip-column-names -h {host} -P {port} -u {user} -e {query}'
            stdin = database_server.password()
        stdout = runner.execute_command(
            command,
            dict(
                mysql=database_server.mysql_bin,
                user=database_server.user(), password=database_server.password(),
                host=database_server.host(), port=database_server.port(),
                query='SELECT user FROM mysql.user'
            ),
            ExecutionOptions(stdin_content=stdin),
        ).stdout

        return set(split_nonempty_strs(stdout))


def get_mssql_host_for_target_server(source_database_server):
    r"""Get hostname of MSSQL database server that could be used on target server
    This function replaces all references to the source server in hostname. So, for example if
    name of a server was 'localhost\MSSQLSERVER2008', it will be replaced with '10.52.1.17\MSSQLSERVER2008',
    where 10.52.1.17 is an IP address of the source database server. Named and default MSSQL instances are handled
    by this function
    :type source_database_server: parallels.core.connections.database_servers.base.DatabaseServer
    """
    hostname, instance, port = split_mssql_host_parts(source_database_server)
    if hostname in ('localhost', '.', '127.0.0.1'):
        hostname = source_database_server.ip()

    return join_mssql_hostname_and_instance(hostname, instance, port)


@synchronized
@cached
def are_same_mssql_servers(source_database_server, target_database_server):
    """Check if two MSSQL servers are actually the same server

    First, we try to check if servers are different by comparing hostname, instance name and port.
    If these items are different (considering resolving of hostname to IP addresses),
    then the servers are different.

    Checking by hostname and instance name does not work in all cases. In situation when
    different servers are attached by private IP address, and this IP address
    is the same, we will consider these servers different, which is not correct. Example:
    [Plesk server #1 (1.2.3.4)] <----- [Database server #1 (10.58.1.15\INSTANCE1)]
    [Plesk server #2 (5.6.7.8)] <----- [Database server #2 (10.58.1.15\INSTANCE1)]
    (here we consider that 10.58.* is a private network)

    So, if hosname, instance name and port are the same, we continue checks by creating
    temporary object (database) on the target server and checking if it appeared on the source server.
    If it appeared - these are the same servers, otherwise these are different servers.

    Also, there is another possible way: checking by some unique identifier, for example date of creation of "master"
    database. But there could be problems in case of multiple MSSQL servers cloned from one virtual disk image.

    :type source_database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :type target_database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: bool
    """
    logger.fdebug(
        messages.DEBUG_CHECK_IF_SERVERS_ARE_SAME,
        source_server=source_database_server.description(),
        target_server=target_database_server.description()
    )
    # Step #1: try to check by simple comparison of hostname, instance name and port
    if not _are_same_mssql_servers_by_connection_data(source_database_server, target_database_server):
        # different hostname, instance name or port means these are guaranteed to be different servers
        logger.fdebug(
            messages.DEBUG_DIFFERENT_SERVERS,
            source_server=source_database_server.description(),
            target_server=target_database_server.description()
        )
        return False

    # Step #2: try to check by creating temporary database table
    same = _are_same_mssql_servers_by_creating_new_object(source_database_server, target_database_server)
    if same:
        logger.fdebug(
            messages.DEBUG_SAME_SERVER,
            source_server=source_database_server.description(),
            target_server=target_database_server.description()
        )
    else:
        logger.finfo(
            messages.DEBUG_DIFFERENT_SERVERS,
            source_server=source_database_server.description(),
            target_server=target_database_server.description()
        )
    return same


def _are_same_mssql_servers_by_connection_data(source_database_server, target_database_server):
    """Check if two MSSQL servers are actually the same server by checking hostname, instance name and port

    :type source_database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :type target_database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: bool
    """
    hostname_1, instance_1, port_1 = split_mssql_host_parts(source_database_server)
    hostname_2, instance_2, port_2 = split_mssql_host_parts(target_database_server)
    # if port is not specified via database server host string, retrieve it from database server info
    if port_1 is None:
        port_1 = _normalize_mssql_port(source_database_server)
    if port_2 is None:
        port_2 = _normalize_mssql_port(target_database_server)

    same_host = (
        source_database_server.ip() in _resolve_mssql_hostname(target_database_server) or
        target_database_server.ip() in _resolve_mssql_hostname(source_database_server)
    )
    same_instance = instance_1 == instance_2
    same_port = port_1 == port_2

    return same_host and same_instance and same_port


def _are_same_mssql_servers_by_creating_new_object(source_database_server, target_database_server):
    """Check if two MSSQL servers are actually the same server by creating new object on one of them

    :type source_database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :type target_database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: bool
    """
    max_attempts = 100  # we will try to find unused name 100 times, it should be enough in all cases
    database_name = None
    for attempt in range(max_attempts):
        random_digits = "".join(random.choice(string.digits) for _ in range(20))
        random_database_name = 'plesk_migrator_%s' % random_digits
        if (
            not _mssql_database_exists(source_database_server, random_database_name) and
            not _mssql_database_exists(target_database_server, random_database_name)
        ):
            database_name = random_database_name
            break

    if database_name is None:
        logger.ferror(
            messages.ERROR_FAILED_TO_FIND_UNUSED_DATABASE_NAME,
            source_server=source_database_server.description(), target_server=target_database_server.description()
        )
        return False

    _mssql_create_empty_database(target_database_server, database_name)
    try:
        same_servers = _mssql_database_exists(source_database_server, database_name)
    finally:
        _mssql_drop_database(target_database_server, database_name)
    return same_servers


def _mssql_database_exists(database_server, database_name):
    """Check whether database with specified name exists on MSSQL server

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :type database_name: str | unicode
    """
    with database_server.panel_server.runner() as runner:
        query_results = runner.mssql_query(
            MSSQLConnectionSettings.create_from_server(database_server),
            "SELECT COUNT(*) AS cnt FROM master.dbo.sysdatabases WHERE name = %(database_name)s",
            dict(database_name=database_name)
        )
        if len(query_results) > 0 and 'cnt' in query_results[0]:
            return int(query_results[0]['cnt']) > 0
        else:
            return False


def _mssql_create_empty_database(database_server, database_name):
    """Create empty database on MSSQL server

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :type database_name: str | unicode
    """
    with database_server.panel_server.runner() as runner:
        runner.mssql_execute(
            MSSQLConnectionSettings.create_from_server(database_server),
            "CREATE DATABASE %s" % database_name
        )


def _mssql_drop_database(database_server, database_name):
    """Drop database from MSSQL server

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :type database_name: str | unicode
    """
    with database_server.panel_server.runner() as runner:
        runner.mssql_execute(
            MSSQLConnectionSettings.create_from_server(database_server),
            "DROP DATABASE %s" % database_name
        )


def _resolve_mssql_hostname(database_server):
    """Convert MSSQL database server hostname (without instance name) to list of IP addresses

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    """
    hostname, _, _ = split_mssql_host_parts(database_server)
    if hostname in ('localhost', '.', '127.0.0.1'):
        return [database_server.ip()]
    else:
        with database_server.panel_server.runner() as runner:
            return runner.resolve_all(hostname)


def split_mssql_host_parts(database_server):
    """Split MSSQL database servers hostname into host and instance parts

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: tuple(str | unicode, str | unicode | None, int | None)
    """
    return split_mssql_host_parts_str(database_server.host())


def split_mssql_host_parts_str(host, can_be_incorrect=False):
    """Split MSSQL database servers hostname into host and instance parts,
    supported hostname format is: <host_name>\<instance_name>,<port>

    :type can_be_incorrect: bool
    :type host: str | unicode
    :rtype: tuple(str | unicode, str | unicode | None, int | None) | None
    """
    # remove port and split rest part on host name and instance name
    parts = host.split(',')
    port = int(parts[1]) if len(parts) > 1 else None
    parts = parts[0].split('\\')
    if len(parts) == 1:
        # Plain hostname
        return parts[0], None, port
    elif len(parts) == 2:
        instance_name = parts[1]
        if instance_name == '':
            instance_name = None
        return parts[0], instance_name, port
    else:
        if can_be_incorrect:
            logger.fdebug(messages.ASSERT_INVALID_MSSQL_HOSTNAME, hostname=host)
            return None
        else:
            raise MigrationNoRepeatError(safe_format(messages.ASSERT_INVALID_MSSQL_HOSTNAME, hostname=host))


def join_mssql_hostname_and_instance(hostname, instance=None, port=None):
    """Join hostname and instance name for MSSQL

    Simply join host and instance name into a single MSSQL hostname.
    Situation when instance name is not set is handled - just hostname is returned in that case.

    :type hostname: str | unicode
    :type instance: str | unicode | None
    :type port: str | unicode | None
    :rtype: str | unicode
    """
    result = hostname
    if not is_empty(instance):
        result = r'%s\%s' % (result, instance)
    if not is_empty(port):
        result = r'%s,%s' % (result, port)
    return result


def _normalize_mssql_port(database_server):
    """Get valid integer MSSQL server port, considering that all null values mean default port (1433)

    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: int
    """
    if database_server.port() in (None, '', 0):
        return 1433
    else:
        return database_server.port()


class MySQLCapabilities(Entity):
    def __init__(self, has_routines, has_events):
        """
        :type has_routines: bool
        :type has_events: bool
        """
        self._has_routines = has_routines
        self._has_events = has_events

    @property
    def has_routines(self):
        """
        :rtype: bool
        """
        return self._has_routines

    @property
    def has_events(self):
        """
        :rtype: bool
        """
        return self._has_events


@cached
def _get_mysqldump_capabilities(server):
    """Get mysqldump utility capabilities - features which could be dumped

    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: parallels.core.utils.database_utils.MySQLCapabilities
    """
    with server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        if server.is_windows():
            help = runner.execute_command(
                '{path_to_mysqldump} --help',
                dict(
                    path_to_mysqldump=server.panel_server.get_path_to_mysqldump(),
                )
            ).stdout
        else:
            help = runner.execute_command('mysqldump --help').stdout

    return MySQLCapabilities(
        has_routines='--routines' in help,
        has_events='--events' in help
    )


def db_type_pretty_name(dbtype):
    """Get pretty human-readable name for database type constant

    For example, 'mysql' constant is converted to "MySQL".

    :type dbtype: str | unicode
    :rtype: str | unicode
    """
    db_server_types = {
        DatabaseServerType.MYSQL: 'MySQL',
        DatabaseServerType.POSTGRESQL: 'PostgreSQL',
        DatabaseServerType.MSSQL: 'MSSQL'
    }
    return db_server_types.get(dbtype, dbtype)


def is_mysql_secure_auth_enabled(server):
    """Check whether secure auth option is enabled for specified MySQL server

    Return True if it is enabled, False if it is disabled, None if we were unable to detect value.

    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: boolean | None
    """
    secure_auth_value = get_mysql_variable(server, 'secure_auth')
    if secure_auth_value is None:
        return None
    else:
        return secure_auth_value.lower() == 'on'


def is_mysql_innodb_strict_mode(server):
    """Get innodb_strict_mode variable for specified MySQL server

    Return True if it is set to 1, False if it is set to 0, None if we were unable to detect value.

    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: boolean | None
    """
    innodb_strict_mode = get_mysql_variable(server, 'innodb_strict_mode')
    if innodb_strict_mode is None:
        return None
    else:
        return innodb_strict_mode.lower() == 'on'


def get_mysql_max_allowed_packet_size(server):
    """Get value of maximum allowed packet size for MySQL server

    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: int | None
    """
    max_allowed_packet_size = get_mysql_variable(server, 'max_allowed_packet')
    if max_allowed_packet_size is None:
        return None
    else:
        return int(max_allowed_packet_size)


def get_mysql_variable(server, variable_name):
    """Get value of MySQL system variable

    :type server: parallels.core.connections.database_servers.base.DatabaseServer
    :type variable_name: str | unicode
    :rtype: str | unicode | None
    """
    query = u"SHOW VARIABLES LIKE '%s'" % variable_name

    if server.is_windows:
        command = (
            u'{mysql_bin} -h {host} -P {port} -u{login} -p{password}'
            u' --silent --skip-column-names -e {query}'
        )
        args = dict(
            mysql_bin=server.mysql_bin, host=server.host(), port=server.port(),
            login=server.user(), password=server.password(), query=query
        )
    else:
        options_file = _create_mysql_options_file(server, 'secure_auth_check')
        command = (
            u"mysql --defaults-file={options_file} --silent --skip-column-names -h {host} -P {port} -e {query}"
        )
        args = dict(
            options_file=options_file, host=server.host(), port=server.port(), query=query
        )
    with server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        query_result = runner.execute_command(command, args).stdout

    parts = query_result.split(None, 1)

    if len(parts) == 2:
        return parts[1].strip()
    else:
        return None


def is_local_database_host(host, server_type, local_ips):
    """Check if specified database host name is local to server with specified IPs

    :type host: str | unicode
    :type server_type: str | unicode
    :type local_ips: list[str | unicode]
    """
    if server_type == DatabaseServerType.MSSQL:
        host, _, _ = split_mssql_host_parts_str(host)
        if host.lower() in ['.', '', '(local)']:
            return True
    if host.lower() == 'localhost':
        return True

    def is_ip_address_local(ip):
        if ip in ['127.0.0.1', '::1']:
            return True
        if ip in local_ips:
            return True
        return False

    if is_ip_address_local(host):
        return True
    for ip_address in resolve_all_safe(host):
        if is_ip_address_local(ip_address):
            return True
    return False


@cached
def _mysql_supports_utf8mb4(database_server):
    """
    :type database_server: parallels.core.connections.database_servers.base.DatabaseServer
    :rtype: bool | None
    """
    if database_server.type() != DatabaseServerType.MYSQL:
        return None

    logger.fdebug(messages.LOG_CHECK_UTFMB4_SUPPORT)
    with database_server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        args = dict(
            user=database_server.user(),
            host=database_server.host(),
            port=database_server.port(),
            query='SHOW CHARACTER SET like \'utf8mb4\''
        )
        if isinstance(runner, WindowsRunner):
            command = '{mysql} --silent --skip-column-names -h {host} -P {port} -u {user} -p{password} -e {query}'
            args.update(
                mysql=database_server.panel_server.get_path_to_mysql(),
                password=database_server.password(),
            )
            stdin = None
        else:
            command = (
                'MYSQL_PWD="$(cat)" mysql --silent --skip-column-names -h {host} -P {port} -u {user} -e {query}'
            )
            stdin = database_server.password()
        execution_options = ExecutionOptions(stdin_content=stdin, ignore_exit_code=True)

        reply = runner.execute_command(command, args, execution_options)

        if reply.exit_code != 0:
            logger.fdebug(messages.LOG_CHECK_UTFMB4_SUPPORT_FAILED)
            return None

        if len(reply.stdout.strip()) == 0:
            logger.fdebug(messages.LOG_UTFMB4_NOT_SUPPORTED)
            return False

        logger.fdebug(messages.LOG_UTFMB4_SUPPORTED)
        return True
