"""
Single interface for running commands via different transports (for example, SSH, Windows HCL, Unix HCL).
"""

from parallels.core import messages
import logging
import pipes
import os
import shutil
import ntpath
from contextlib import closing
import errno
from parallels.core.utils.zip_utils import zip_directory

from parallels.core.utils.common import poll_data, polling_intervals, safe_string_repr, strip_multiline_spaces,\
	split_nonempty_strs, cached, mkdir_p, get_env_str, default
from parallels.core.registry import Registry
from parallels.core.utils.steps_profiler import sleep
from parallels.core.utils import windows_utils, ssh_utils, unix_utils
from parallels.core.utils import windows_thirdparty
from parallels.core import local_command_exec
from parallels.core import MigrationError
from parallels.core.utils.windows_agent_pool import WindowsAgentPool
from parallels.core.utils.windows_utils import check_windows_national_symbols_command
from parallels.core.utils.steps_profiler import get_default_steps_profiler
from parallels.core.utils.common.logging import PositionalArgCommand, NamedArgCommand, hide_text
from parallels.core.utils.common.threading_utils import LocksByKey

logger = logging.getLogger(__name__)
profiler = get_default_steps_profiler()


class SSHRunnerException(MigrationError):
	def __init__(self, message, command_str=None, stdout=None, stderr=None):
		super(MigrationError, self).__init__(message)
		self.command_str = command_str
		self.stdout = stdout
		self.stderr = stderr


class BaseRunnerException(MigrationError):
	def __init__(self, message):
		super(MigrationError, self).__init__(message)


class ConnectionCheckError(MigrationError):
	pass


class WinexeRunnerException(MigrationError):
	def __init__(self, message, command_str=None, full_command_str=None, stdout=None, stderr=None):
		super(MigrationError, self).__init__(message)
		self.command_str = command_str
		self.full_command_str = full_command_str
		self.stdout = stdout
		self.stderr = stderr


class BaseRunner(object):
	def run_unchecked(self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict', env=None):
		"""Run command with specified args. Arguments are properly escaped."""
		raise NotImplementedError()

	def sh_unchecked(
		self, cmd_str, args=None, stdin_content=None, output_codepage=None,
		error_policy='strict', env=None, log_output=True
	):
		"""Substitute format fields in 'cmd_str', then execute the command string.
		
		For Unix: whole command is executed, then if you run

			"runner.sh('ls *')"

		, '*' will be substituted by shell, while if you run

			"runner.sh('ls {file}', dict(file='*'))"

		, '*' will be escaped, much like "run('ls', ['*'])".
		
		For Windows: there are no general command arguments parsing rules, and
		'*' is subsitituded by each particular program, this function differs
		from run_unchecked only by a way to substitute arguments.

		If None is passed as args, no command formatting is performed, command
		run as is.

		Returns a tuple (exit_code, stdout, stderr)
		"""
		raise NotImplementedError()

	def run(self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict', env=None):
		"""The same as run_unchecker(), but checks exit code and returns only stdout"""
		exit_code, stdout, stderr = self.run_unchecked(cmd, args, stdin_content, output_codepage, error_policy, env)
		if exit_code != 0:
			raise BaseRunnerException(
				messages.RUNNER_COMMAND_FAILED_WITH_EXIT_CODE % (
					cmd, args, exit_code, safe_string_repr(stdout), safe_string_repr(stderr)
				)
			)
		return stdout 

	def sh(
		self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict',
		env=None, log_output=True
	):
		"""The same as sh_unchecked(), but checks exit code and returns only stdout"""
		exit_code, stdout, stderr = self.sh_unchecked(cmd, args, stdin_content, output_codepage, error_policy, env)
		if exit_code != 0:
			raise BaseRunnerException(
				messages.RUNNER_COMMAND_FAILED_WITH_EXIT_CODE % (
					cmd, args, exit_code,
					safe_string_repr(hide_text(stdout, not log_output)),
					safe_string_repr(hide_text(stderr, not log_output))
				)
			)
		return stdout

	def sh_multiple_attempts(
		self, cmd_str, args=None, stdin=None, codepage=None,
		error_policy='strict', env=None, max_attempts=5,
		interval_between_attempts=10, log_output=True
	):
		"""Run command with several attempts; check exit code; return only stdout."""
		from parallels.core.utils.migrator_utils import get_bleeping_logger
		blogger = get_bleeping_logger(logger)
		command = self._get_command_object(cmd_str, args, self._get_command_str)
		for attempt in range(0, max_attempts):
			blogger.debug(messages.COMMAND_RUN_ATTEMPTS, command)
			exit_code, stdout, stderr = self.sh_unchecked(
				cmd_str, args, stdin, codepage, error_policy, env, log_output=log_output
			)
			if exit_code == 0:
				if attempt > 0:
					blogger.info(messages.COMMAND_EXECUTED_OK_AFTER_ATTEMPTS, command, attempt+1)
				return stdout
			elif attempt >= max_attempts - 1:
				failed_cmd = blogger.format('%s', command)
				self._raise_runner_exception(failed_cmd, stdout, stderr)
			else:
				blogger.error(messages.FAILED_EXECUTE_REMOTE_COMMAND_RETRY_IN % interval_between_attempts)
				sleep(interval_between_attempts, messages.SLEEP_RETRY_EXECUTING_COMMAND)

	@staticmethod
	def _get_command_object(command, args, serializer):
		if isinstance(args, dict):
			command_class = NamedArgCommand
		elif isinstance(args, list):
			command_class = PositionalArgCommand
		else:
			if args is None:
				return command
			else:
				return '%s %s' % (command, args)

		return command_class(command, args, serializer)

	def _raise_runner_exception(self, cmd_str, stdout, stderr):
		raise NotImplementedError()

	def upload_file_content(self, filename, content):
		raise NotImplementedError()

	def upload_file(self, local_filename, remote_filename):
		raise NotImplementedError()

	def get_file(self, remote_filename, local_filename):
		raise NotImplementedError()

	def get_file_contents(self, remote_filename):
		raise NotImplementedError()

	def move(self, src_path, dst_path):
		raise NotImplementedError()

	def remove_file(self, filename):
		"""Remove file. If file does not exist - silently do nothing.

		:type filename: basestring
		"""
		raise NotImplementedError()

	def remove_directory(self, directory, is_remove_root=True):
		raise NotImplementedError()

	def clean_directory(self, directory):
		self.remove_directory(directory, False)

	def file_exists(self, filename):
		raise NotImplementedError()

	def mkdir(self, dirname):
		raise NotImplementedError()

	def download_directory(self, remote_directory, local_directory):
		raise NotImplementedError()

	def download_directory_as_zip(self, remote_directory, local_zip_filename):
		raise NotImplementedError()

	def check(self, server_description):
		"""Check connection with that runner

		:type server_description: basestring
		:rtype None
		"""
		raise NotImplementedError()

	def get_files_list(self, path):
		raise NotImplementedError()

	@staticmethod
	def _get_command_str(cmd_str, args=None):
		raise NotImplementedError()


class WindowsRunner(BaseRunner):
	"""Add codepage detection to migrator's Windows CMD interface classes."""
	@property
	@cached
	def codepage(self):
		"""Determine code page used by Windows CMD.

		'CHCP' utility returns Windows codepage, for example:
		"Active code page: 123"

		Python codec name is derived from 'chcp' output by adding 'cp' to code
		page number, for example: '866' -> 'cp866'
		
		Returns:
		    Python codec name to be used for decoding CMD output.
		"""

		code, stdout, stderr = self._run_command_and_decode_output(
				"cmd.exe /c CHCP", 'ascii', error_policy='ignore')
		if code != 0:
			raise BaseRunnerException(
				messages.UNABLE_DETECT_WINDOWS_CODEPAGE % (
					safe_string_repr(stdout), safe_string_repr(stderr)
				)
			)
		else:
			codepage =  u"cp%s" % stdout.split()[-1].strip('.')
			logger.debug(messages.DEBUG_DETECTED_CODEPAGE % codepage)

		return codepage

	def move(self, src_path, dst_path):
		self.sh('cmd /c move {src_path} {dst_path}', dict(
			src_path=src_path,
			dst_path=dst_path
		))

	def get_files_list(self, path):
		output = self.sh('cmd /c dir /b {path}', dict(path=path))
		return split_nonempty_strs(output.strip(), "\n")

	def remove_file(self, filename):
		self.sh('cmd /c del {filename}', dict(filename=filename))

	def remove_directory(self, directory, is_remove_root=True):
		# remove all files in a directory recursively
		self.sh('cmd /c del {directory} /s /q', dict(directory=directory))
		# remove all directories recursively
		directories_to_delete = directory if is_remove_root else ntpath.join(directory, '*')
		self.sh(
			u'cmd /c for /d %p in ({directories_to_delete}) do rmdir "%p" /s /q',
			dict(directories_to_delete=directories_to_delete)
		)

	def _run_command_and_decode_output(self, cmd_str, output_codepage=None, error_policy='strict', env=None):
		"""Run a command, return its output decoded to UTF."""
		raise NotImplementedError()

	def file_exists(self, filename):
		return windows_utils.file_exists(self, filename)


class LocalUnixRunner(BaseRunner):
	"""Execute commands on local server"""

	@property
	def codepage(self):
		return 'utf-8'

	def run_unchecked(self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict', env=None):
		command = [cmd] + ([] if args is None else args)
		codepage = output_codepage if output_codepage else self.codepage
		return local_command_exec(command, stdin_content, codepage, error_policy, env)

	def sh_unchecked(
		self, cmd_str, args=None, stdin_content=None, output_codepage=None,
		error_policy='strict', env=None, log_output=True
	):
		if args is None:
			command = ['/bin/sh', '-c', cmd_str]
		else:
			formatted_cmd = unix_utils.format_command(cmd_str, **args)
			command = ['/bin/sh', '-c', formatted_cmd]
		codepage = output_codepage if output_codepage else self.codepage
		return local_command_exec(command, stdin_content, codepage, error_policy, env, log_output=log_output)

	def mkdir(self, dirname):
		return self.sh('mkdir -p {dirname}', dict(dirname=dirname))

	def upload_file_content(self, filename, content):
		with open(filename, "wb") as fp:
			fp.write(content)

	def upload_file(self, local_filename, remote_filename):
		if local_filename != remote_filename:
			shutil.copy2(local_filename, remote_filename)

	def get_file(self, remote_filename, local_filename):
		if local_filename != remote_filename:
			shutil.copy2(remote_filename, local_filename)

	def get_file_contents(self, remote_filename):
		with open(remote_filename, "rb") as fp:
			return fp.read().decode('utf-8')

	def move(self, src_path, dst_path):
		shutil.move(src_path, dst_path)

	def get_files_list(self, path):
		return os.listdir(path)

	def remove_file(self, filename):
		os.remove(filename)

	def check(self, server_description):
		# No need to check local execution, it should never fail
		pass

	def file_exists(self, filename):
		return os.path.exists(filename)


class LocalWindowsRunner(WindowsRunner):
	"""Execute commands on local Windows server"""

	def __init__(self, disable_profiling=False):
		self._disable_profiling = disable_profiling

	@staticmethod
	@cached
	def get_instance(disable_profiling=False):
		return LocalWindowsRunner(disable_profiling)

	def _run_command_and_decode_output(self, cmd_str, output_codepage=None, error_policy='strict', env=None):
		codepage = output_codepage if output_codepage else self.codepage
		return local_command_exec(cmd_str, None, codepage, error_policy, env)

	def sh_unchecked(
		self, cmd_str, args=None, stdin_content=None, output_codepage=None,
		error_policy='strict', env=None, log_output=True
	):
		if args is not None:
			cmd_str = windows_utils.format_command(cmd_str, **args)
		logger.debug(messages.EXECUTE_COMMAND_ON_LOCAL_SERVER, cmd_str)
		codepage = output_codepage if output_codepage else self.codepage

		if not self._disable_profiling:
			with profiler.measure_command_call(cmd_str, messages.LOCAL_SERVER_TITLE):
				return local_command_exec(cmd_str, stdin_content, codepage, error_policy, env)
		else:
			return local_command_exec(cmd_str, stdin_content, codepage, error_policy, env)

	def run_unchecked(self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict', env=None):
		command = windows_utils.format_command_list(cmd, args)
		args = None
		return self.sh_unchecked(command, args, stdin_content, output_codepage, error_policy, env)

	def upload_file(self, local_filename, remote_filename):
		shutil.copy2(local_filename, remote_filename)

	def upload_file_content(self, filename, content):
		with open(filename, "wb") as fp:
			fp.write(content)

	def get_file(self, remote_filename, local_filename):
		if local_filename != remote_filename:
			logger.debug(messages.DEBUG_COPY_LOCAL_FILE, remote_filename, local_filename)
			shutil.copy2(remote_filename, local_filename)

	def get_file_contents(self, remote_filename):
		with open(remote_filename, 'rb') as fp:
			return fp.read().decode('utf-8')

	def move(self, src_path, dst_path):
		shutil.move(src_path, dst_path)

	def get_files_list(self, path):
		return os.listdir(path)

	def mkdir(self, dirname):
		self.sh('cmd /c if not exist {dirname} ( md {dirname} )', dict(dirname=dirname))

	def remove_file(self, filename):
		try:
			os.remove(filename)
		except OSError as e:
			# if file does not exist - just skip
			if e.errno != errno.ENOENT:
				raise

	def remove_directory(self, directory, is_remove_root=True):
		for root, dirs, files in os.walk(directory):
			for name in files:
				self.remove_file(ntpath.join(directory, name))
			for name in dirs:
				self.remove_directory(ntpath.join(directory, name))
			if is_remove_root:
				os.rmdir(root)

	def check(self, server_description):
		# No need to check local execution, it should never fail
		pass


class SSHRunner(BaseRunner):
	def __init__(self, ssh, host_description=None):
		self.ssh = ssh
		self.host_description = host_description

	@property
	def codepage(self):
		return 'utf-8'

	def run(self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict', env=None):
		"""The same as run_unchecked(), but checks exit code and returns only stdout"""
		exit_code, stdout, stderr = self.run_unchecked(cmd, args, stdin_content, output_codepage, error_policy, env)
		if exit_code != 0:
			self._raise_runner_exception(unix_utils.format_command_list(cmd, args), stdout, stderr)
		return stdout 

	def run_unchecked(self, cmd, args=None, stdin_content=None, codepage=None, error_policy='strict', env=None):
		"""Execute a command with a list of positional 'args'."""
		formatter = unix_utils.format_command_list
		return self._exec(cmd, args, formatter, stdin_content, codepage, error_policy, env)

	def sh_unchecked(
		self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict',
		env=None, log_output=True
	):
		"""Format and execute a command with a dict of named 'args'."""
		formatter = self._get_command_str
		return self._exec(
			cmd, args, formatter, stdin_content, output_codepage, error_policy, env, log_output=log_output
		)

	def _exec(self, cmd, args, formatter, stdin_content, output_codepage, error_policy, env, log_output=True):
		self._log_command(cmd, args, formatter)
		codepage = output_codepage if output_codepage else self.codepage
		command_str = formatter(cmd, args)
		with profiler.measure_command_call(command_str, self.host_description):
			env = default(env, {})
			# Execute all commands under 'en_US.utf-8' locale, if it is not specified,
			# so migrator works in the same way regardless of server's locale
			if 'LANG' not in env:
				env['LANG'] = 'en_US.utf-8'

			return ssh_utils.run_unchecked(
				self.ssh, command_str, stdin_content, codepage, error_policy, env, log_output=log_output
			)

	def _log_command(self, cmd, args, formatter):
		"""Write command string to log while omitting sensitive data."""
		from parallels.core.utils.migrator_utils import get_bleeping_logger
		blogger = get_bleeping_logger(logger)
		if isinstance(args, dict):
			command = NamedArgCommand(cmd, args, formatter)
		else:
			command = PositionalArgCommand(cmd, args, formatter)
		blogger.debug(messages.DEBUG_EXECUTE_SSH_REMOTE_COMMAND, self.ssh.hostname, command)

	def sh(
		self, cmd_str, args=None, stdin_content=None, output_codepage=None, error_policy='strict',
		env=None, log_output=True
	):
		"""The same as sh_unchecked(), but checks exit code and returns only stdout"""
		return self.sh_multiple_attempts(
			cmd_str, args, stdin_content, output_codepage, error_policy, env,
			max_attempts=1, log_output=log_output
		)

	def upload_file(self, local_filename, remote_filename):
		with closing(self.ssh.open_sftp()) as sftp:
			sftp.put(local_filename, remote_filename)

	def upload_file_content(self, filename, content):
		with closing(self.ssh.open_sftp()) as sftp:
			with closing(sftp.open(filename, 'w')) as fp:
				fp.write(content)

	def get_file(self, remote_filename, local_filename):
		with closing(self.ssh.open_sftp()) as sftp:
			sftp.get(remote_filename, local_filename)

	def get_file_contents(self, remote_filename):
		return self.run('/bin/cat', [remote_filename])

	def move(self, src_path, dst_path):
		return self.run('/bin/mv', [src_path, dst_path])

	def get_files_list(self, path):
		return unix_utils.get_files_list(self, path)

	def remove_file(self, filename):
		return self.run('/bin/rm', [filename])

	def mkdir(self, dirname):
		return self.run('/bin/mkdir', ['-p', dirname])

	def mktempdir(self, dirname):
		"""Create a 'dirname' subdirectory with unique name."""
		self.mkdir(dirname)
		tmpdirname = self.run('/bin/mktemp', ['-d', '-p', dirname])
		return tmpdirname.strip()

	def _raise_runner_exception(self, command, stdout, stderr):
		if self.host_description is not None:
			where = " " + messages.EXECUTED_AT % (self.host_description,)
		else:
			where = ""

		raise SSHRunnerException(
			messages.COMMAND_COMMANDWHERE_RETURNED_NONZERO_EXIT_CODE.format(
				command=command, where=where, stderr=safe_string_repr(stderr), stdout=safe_string_repr(stdout)
			),
			command_str=command,
			stdout=stdout,
			stderr=stderr
		)

	@staticmethod
	def _get_command_str(cmd_str, args=None):
		"""Format command string with named formatting arguments."""
		if args is not None and len(args) > 0:
			cmd_str = unix_utils.format_command(cmd_str, **args)
		return cmd_str

	def file_exists(self, filename):
		return unix_utils.file_exists(self, filename)


class WindowsAgentRunner(BaseRunner):
	"""
	Runner for Windows nodes which have server based on
	lib/python/parallels/source/common/extras/transfer-agent/server.py

	Runner tries to install Windows agent automatically on remote node with paexec.
	In case that is not possible (for example, there is no Samba connection, so
	paexec does not work), you could install agent on remote node manually.
	"""
	def __init__(self, settings, migrator_server, host_description=None):
		self.remote = WindowsAgentPool.get_instance(migrator_server).get(settings)
		self.settings = settings
		if host_description is not None:
			self.host_description = settings.ip
		else:
			self.host_description = host_description
		self.host_description = settings.ip

	def check(self, server_description):
		try:
			self.sh('cmd /c echo 1')
		except Exception:
			logger.debug(messages.LOG_EXCEPTION, exc_info=True)
			raise ConnectionCheckError(
				messages.FAILED_TO_CONNECT_TO_SERVER_WITH_TRANSFER_AGENT.format(server=server_description)
			)

	def sh_unchecked(
		self, cmd_str, args=None, stdin_content=None, output_codepage=None,
		error_policy='strict', env=None, log_output=True
	):
		if args is not None:
			cmd_str_formatted = windows_utils.format_command(cmd_str, **args)
		else:
			cmd_str_formatted = cmd_str

		logger.debug(messages.DEBUG_EXECUTE_COMMAND_AT, self.settings.ip, cmd_str_formatted)
		with profiler.measure_command_call(cmd_str_formatted, self.host_description):
			exit_code, stdout, stderr = self.remote.sh_unchecked(cmd_str, args, stdin_content, env)

		logger.debug(messages.DEBUG_COMMAND_STDOUT, stdout)
		logger.debug(messages.DEBUG_COMMAND_STDERR, stderr)
		logger.debug(messages.DEBUG_COMMAND_EXIT_CODE, exit_code)

		return exit_code, stdout, stderr

	def run_unchecked(self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict', env=None):
		command = windows_utils.format_command_list(cmd, args)
		args = None
		return self.sh_unchecked(command, args, stdin_content, output_codepage, error_policy, env)

	def get_file(self, remote_filename, local_filename):
		logger.debug(messages.DEBUG_DOWNLOAD_FILE, remote_filename, local_filename, self.settings.ip)
		with open(local_filename, 'wb') as fp:
			fp.write(self.get_file_contents(remote_filename))

	def upload_file(self, local_filename, remote_filename):
		logger.debug(messages.DEBUG_UPLOAD_FILE, local_filename, remote_filename, self.settings.ip)
		with open(local_filename, 'rb') as fp:
			self.upload_file_content(remote_filename, fp.read())

	def upload_file_content(self, filename, content):
		logger.debug(messages.DEBUG_UPLOAD_FILE_CONTENTS, filename, self.settings.ip)
		with profiler.measure_command_call(messages.PROFILE_UPLOAD_FILE % (filename,), self.host_description):
			self.remote.upload_file_content(filename, content)

	def get_file_contents(self, remote_filename):
		logger.debug(messages.DEBUG_DOWNLOAD_FILE_CONTENTS, remote_filename, self.settings.ip)
		with profiler.measure_command_call(messages.PROFILE_DOWNLOAD_FILE % (remote_filename,), self.host_description):
			return self.remote.get_file_contents(remote_filename)

	def move(self, src_path, dst_path):
		logger.debug(messages.DEBUG_MOVE_FILE % (src_path, dst_path, self.settings.ip))
		with profiler.measure_command_call(messages.PROFILE_MOVE_FILE % (src_path, dst_path), self.host_description):
			self.remote.move(src_path, dst_path)

	def get_files_list(self, path):
		logger.debug(messages.DEBUG_LIST_FILES % (path, self.settings.ip))
		with profiler.measure_command_call(messages.PROFILE_LIST_FILES % path, self.host_description):
			return self.remote.get_files_list(path)

	def remove_file(self, filename):
		logger.debug(messages.DEBUG_REMOVE_FILE, filename, self.settings.ip)
		with profiler.measure_command_call(messages.PROFILE_REMOVE_FILE % filename, self.host_description):
			self.remote.remove_file(filename)

	def mkdir(self, dirname):
		logger.debug(messages.DEBUG_CREATE_DIRECTORY, dirname, self.settings.ip)
		with profiler.measure_command_call(messages.PROFILE_CREATE_DIRECTORY % (dirname,), self.host_description):
			self.remote.mkdir(dirname)

	def download_directory(self, remote_directory, local_directory):
		migrator_server = Registry().get_instance().get_context().migrator_server
		with migrator_server.runner() as local_runner:
			local_runner.mkdir(local_directory)
		for directory in self.remote.get_directories_list_recursive(remote_directory):
			with migrator_server.runner() as local_runner:
				local_runner.mkdir(os.path.join(local_directory, directory))

		for filename in self.remote.get_files_list_recursive(remote_directory):
			with migrator_server.runner() as local_runner:
				local_runner.upload_file_content(
					os.path.join(local_directory, filename),
					self.remote.get_file_contents(ntpath.join(remote_directory, filename))
				)

	def download_directory_as_zip(self, remote_directory, local_zip_filename):
		migrator_server = Registry().get_instance().get_context().migrator_server
		with migrator_server.runner() as local_runner:
			local_runner.upload_file_content(local_zip_filename, self.remote.get_directory_as_zip(remote_directory))

	def file_exists(self, filename):
		return self.remote.file_exists(filename)

	def callback(self, command):
		context = Registry.get_instance().get_context()
		target_connection = context.conn.target
		return self.remote.callback(
			command,
			target_connection.main_node_ip,
			target_connection.settings.windows_auth.username,
			target_connection.settings.windows_auth.password
		)

	@staticmethod
	def _get_command_str(cmd_str, args=None):
		return windows_utils.format_command(cmd_str, **args)

	def _raise_runner_exception(self, cmd, stdout, stderr):
		raise BaseRunnerException(messages.FAILED_TO_EXECUTE_COMMAND % (cmd, stdout, stderr))


class PAExecRunner(BaseRunner):
	"""Execute commands on remote Windows server with paexec utility"""

	def __init__(self, settings, migrator_server, source_server):
		"""
		:type settings: parallels.plesk.config.SourcePleskWindowsConfig
		:type migrator_server: parallels.core.connections.migrator_server.MigratorServer
		:type source_server: parallels.core.connections.source_server.SourceServer
		"""
		self.global_context = Registry.get_instance().get_context()
		self.settings = settings
		self.migrator_server = migrator_server
		self.source_server = source_server

	def sh_unchecked(
		self, cmd_str, args=None, stdin_content=None, output_codepage=None,
		error_policy='strict', env=None, log_output=True
	):
		logger.debug(messages.EXECUTE_COMMAND_REMOTE_SERVER_PAEXEC)
		if stdin_content is not None:
			raise NotImplementedError(messages.RUNNER_DOES_NOT_SUPPORT_STDIN_CONTENT_2)

		if env is not None:
			cmd_str = u'%s%s' % (get_env_str(env), cmd_str)

		if args is not None:
			cmd_str = windows_utils.format_command(cmd_str, **args)

		with profiler.measure_command_call(cmd_str, self.settings.ip):
			passwords_file = self.migrator_server.get_session_file_path('node-password-%s' % self.settings.ip)
			with open(passwords_file, 'wb') as f:
				f.write(self.settings.windows_auth.password)
			return LocalWindowsRunner.get_instance(disable_profiling=True).sh_unchecked(
				r'%s \\%s -u %s -p@ %s -p@d ' % (
					windows_thirdparty.get_paexec_bin(),
					self.settings.ip,
					self.settings.windows_auth.username,
					passwords_file	
				) + cmd_str
			)

	def run_unchecked(self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict', env=None):
		command = windows_utils.format_command_list(cmd, args)
		args = None
		return self.sh_unchecked(command, args, stdin_content, output_codepage, error_policy, env)

	def get_file(self, remote_filename, local_filename):
		download_temp_file_basename = 'paexec-download'
		download_temp_file_source_path = self.source_server.get_session_file_path(download_temp_file_basename)
		self.copy(remote_filename, download_temp_file_source_path)
		try:
			rsync = self.global_context.rsync_pool.get(self.source_server, self.migrator_server)
			rsync.sync(
				source_path=rsync.get_rsync_path_to_source_session_dir_file(download_temp_file_basename),
				target_path=windows_utils.convert_path_to_cygwin(local_filename)
			)
		finally:
			self.remove_file(download_temp_file_source_path)

	def get_file_contents(self, remote_filename):
		temp_filename = self.migrator_server.get_session_file_path('paexec-download')
		try:
			self.get_file(remote_filename, temp_filename)
			with open(temp_filename, 'rb') as f:
				return f.read()
		finally:
			if os.path.isfile(temp_filename):
				os.remove(temp_filename)

	def upload_file_content(self, filename, content):
		local_temp_file = self.migrator_server.get_session_file_path('upload')
		with open(local_temp_file, "wb") as fp:
			fp.write(content)
		self.upload_file(local_temp_file, filename)

	def upload_file(self, local_filename, remote_filename):
		upload_temp_file_basename = 'paexec-upload'
		upload_temp_file_source_path = self.source_server.get_session_file_path(upload_temp_file_basename)

		rsync = self.global_context.rsync_pool.get(self.source_server, self.migrator_server)
		rsync.upload(
			source_path=windows_utils.convert_path_to_cygwin(local_filename),
			target_path=rsync.get_rsync_path_to_source_session_dir_file(upload_temp_file_basename)
		)
		self.move(upload_temp_file_source_path, remote_filename)

	def mkdir(self, dirname):
		self.sh('cmd /c if not exist {dirname} ( md {dirname} )', dict(dirname=dirname))

	def move(self, src_path, dst_path):
		self.sh('cmd /c move {src_path} {dst_path}', dict(
			src_path=src_path,
			dst_path=dst_path
		))

	def copy(self, src_path, dst_path):
		"""
		:type src_path: str
		:type dst_path: str
		:rtype: None
		"""
		self.sh('cmd /c copy {src_path} {dst_path}', dict(
			src_path=src_path,
			dst_path=dst_path
		))

	def remove_file(self, filename):
		self.sh('cmd /c del {filename}', dict(filename=filename))

	def remove_directory(self, directory, is_remove_root=True):
		# remove all files in a directory recursively
		self.sh('cmd /c del {directory} /s /q', dict(directory=directory))
		# remove all directories recursively
		directories_to_delete = directory if is_remove_root else ntpath.join(directory, '*')
		self.sh(
			u'cmd /c for /d %p in ({directories_to_delete}) do rmdir "%p" /s /q',
			dict(directories_to_delete=directories_to_delete)
		)

	def file_exists(self, filename):
		return windows_utils.file_exists(self, filename)

	def check(self, server_description):
		try:
			self.sh('cmd /c echo 1')
		except Exception:
			logger.debug(messages.LOG_EXCEPTION, exc_info=True)
			raise ConnectionCheckError(messages.FAILED_CONNECT_SERVER_S_PAEXEC_UTILITY % server_description)


class WinexeRunner(WindowsRunner):
	"""Remotely execute Windows CMD commands on Linux using 'winexe' tool."""

	_locks = LocksByKey()

	def __init__(self, settings, runas=False, local_temp_dir=None):
		self.settings = settings
		self.runas = runas
		self.local_temp_dir = local_temp_dir

	def run_unchecked(self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict', env=None):
		if stdin_content is not None:
			raise NotImplementedError(messages.RUNNER_DOES_NOT_SUPPORT_STDIN_CONTENT_3)

		command_str = windows_utils.format_command_list(cmd, args)
		with profiler.measure_command_call(command_str, self.settings.ip):
			return self._run_command_and_decode_output(command_str, output_codepage, error_policy, env)

	def sh_unchecked(
		self, cmd_str, args=None, stdin_content=None, output_codepage=None,
		error_policy='strict', env=None, log_output=True
	):
		if stdin_content is not None:
			raise NotImplementedError(messages.RUNNER_DOES_NOT_SUPPORT_STDIN_CONTENT_4)

		if args is not None:
			cmd_str = windows_utils.format_command(cmd_str, **args)

		with profiler.measure_command_call(cmd_str, self.settings.ip):
			return self._run_command_and_decode_output(cmd_str, output_codepage, error_policy, env)

	def run(self, cmd, args=None, stdin_content=None, output_codepage=None, error_policy='strict', env=None):
		"""The same as run_unchecked(), but checks exit code and returns only stdout"""
		exit_code, stdout, stderr = self.run_unchecked(cmd, args, stdin_content, output_codepage, error_policy, env)
		if exit_code != 0:
			self._raise_runner_exception(
				windows_utils.format_command_list(cmd, args), stdout, stderr)
		return stdout

	def sh(
		self, cmd_str, args=None, stdin_content=None, output_codepage=None, error_policy='strict',
		env=None, log_output=True
	):
		"""The same as sh_unchecked(), but checks exit code and returns only stdout"""
		return self.sh_multiple_attempts(
			cmd_str, args, stdin_content, output_codepage, error_policy,
			env, max_attempts=1, log_output=log_output
		)

	def _run_command_and_decode_output(self, cmd_str, output_codepage=None, error_policy='strict', env=None):
		"""Run a command, return its output decoded from 'codepage' to UTF."""
		if env is not None:
			cmd_str = u'%s%s' % (get_env_str(env), cmd_str)

		check_windows_national_symbols_command(cmd_str)
		formatted_command = self._construct_winexe_command(cmd_str)
		codepage = output_codepage if output_codepage else self.codepage

		errors = []

		def run_winexe_command():
			with self._locks.lock(self.settings.ip):
				exit_code, stdout, stderr = local_command_exec(formatted_command, None, codepage, error_policy)

			# remove winexe warning with no impact on us
			stderr = stderr.replace(messages.DOS_CHARSET_CP_UNAVAILABLE_USING_ASCII, '')

			if stderr.find(u"Failed to open connection - NT_STATUS_LOGON_FAILURE") != -1:
				errors.append(messages.COMMAND_S_FAILED_EXECUTE_S_ERROR % (
					u" ".join([pipes.quote(c) for c in formatted_command]),
					stderr.strip(),
					self.settings.ip
				))
				return None
			elif stderr.find(u"Failed to open connection") != -1:
				errors.append(messages.COMMAND_S_FAILED_EXECUTE_S_ERROR_1 % (
					u" ".join([pipes.quote(c) for c in formatted_command]),
					stderr.strip(),
					self.settings.ip
				))
				return None
			elif stderr.find(u"NT_STATUS_") != -1:
				errors.append(messages.COMMAND_S_FAILED_EXECUTE_S_ERROR_2 % (
					u" ".join([pipes.quote(c) for c in formatted_command]),
					stderr.strip()
				))
				return None
			else:
				return exit_code, stdout, stderr 
		
		result = poll_data(run_winexe_command, polling_intervals(starting=3, max_attempt_interval=5, max_total_time=30))
		if result is not None:
			return result
		else:
			logger.debug(messages.ERRORS_WHEN_EXECUTING_COMMAND % "\n".join(errors))
			raise MigrationError(errors[len(errors)-1])

	def upload_file(self, local_filename, remote_filename):
		# implementation note: winexe runner utilizes smbclient to upload files
		windows_utils.upload_file(local_filename, remote_filename, self.settings)

	def upload_file_content(self, filename, content):
		if self.local_temp_dir is None:
			raise Exception(messages.UNABLE_UPLOAD_FILE_CONTENTS_S_DIRECTORY % (filename,))
		# implementation note: winexe runner utilizes smbclient to upload files
		temp_filename = os.path.join(self.local_temp_dir, 'winexe_upload')	
		try:
			with open(temp_filename, "w") as f:
				f.write(content)

			windows_utils.upload_file(temp_filename, filename, self.settings)
		finally:
			if os.path.isfile(temp_filename):
				os.remove(temp_filename)

	def get_file(self, remote_filename, local_filename):
		# implementation note: winexe runner utilizes smbclient to download files
		windows_utils.download_file(remote_filename, local_filename, self.settings)

	def get_file_contents(self, remote_filename):
		if self.local_temp_dir is None:
			raise Exception(messages.UNABLE_DOWNLOAD_FILE_CONTENTS_S_DIRECTORY % (remote_filename,))
		temp_filename = os.path.join(self.local_temp_dir, 'winexe_download')
		try:
			windows_utils.download_file(remote_filename, temp_filename, self.settings)
			with open(temp_filename, 'rb') as f:
				return f.read()
		finally:
			if os.path.isfile(temp_filename):
				os.remove(temp_filename)

	def download_directory(self, remote_directory, local_directory):
		with Registry().get_instance().get_context().migrator_server.runner() as local_runner:
			local_runner.mkdir(local_directory)
		windows_utils.download_directory_contents(remote_directory, local_directory, self.settings)

	def download_directory_as_zip(self, remote_directory, local_zip_filename):
		local_temp_directory = os.path.join(self.local_temp_dir, 'download-dirs', os.path.basename(local_zip_filename))
		if os.path.exists(local_temp_directory):
			shutil.rmtree(local_temp_directory)
		mkdir_p(local_temp_directory)
		self.download_directory(remote_directory, local_temp_directory)
		zip_directory(local_temp_directory, local_zip_filename)
		shutil.rmtree(local_temp_directory)

	def mkdir(self, dirname):
		self.sh('cmd /c if not exist {dirname} ( md {dirname} )', dict(dirname=dirname))

	def check(self, server_description):
		try:
			logger.debug(messages.TRY_CONNECT_SAMBA_SHARE_C_S % server_description)
			# no command - just connect to the share and exit
			windows_utils.run_smbclient_cmd("", "C$", self.settings)
		except Exception as err:
			logger.debug(messages.LOG_EXCEPTION, exc_info=True)
			raise ConnectionCheckError(strip_multiline_spaces(
				messages.FAILED_CONNECT_SAMBA_SHARE_C_SERVER % (server_description, err)
			))

	def _construct_winexe_command(self, cmd_str):
		formated_command = [
			u"winexe", u"--debug-stderr", u"--reinstall", u"-U",
			"%s%%%s" % (
				self.settings.windows_auth.username,
				self.settings.windows_auth.password),
			u"//%s" % (self.settings.ip,),
			cmd_str
		]

		if self.runas:
			formated_command += [
				u"--runas", "%s%%%s" % (self.settings.windows_auth.username, self.settings.windows_auth.password)
			]

		return formated_command

	@staticmethod
	def _get_cmd_str_listargs(cmd, args):
		return " ".join([windows_utils.quote_arg(arg) for arg in [cmd] + args])

	@staticmethod
	def _get_cmd_str_formatargs(cmd_str, args):
		if args is not None:
			return windows_utils.format_command(cmd_str, **args)
		else:
			return cmd_str

	def _raise_runner_exception(self, command_str, stdout, stderr):
		full_command = self._construct_winexe_command(command_str)
		full_cmd = full_command[0]
		full_args = full_command[1:]
		full_command_str = unix_utils.format_command_list(full_cmd, full_args)
		where = " " + messages.EXECUTED_AT % (self.settings.ip,)
		raise WinexeRunnerException(
			strip_multiline_spaces(messages.COMMAND_COMMAND_STRWHERE_RETURNED_NONZERO_EXIT).format(
				command_str=command_str, full_command_str=full_command_str,
				where=where, stderr=safe_string_repr(stderr), stdout=safe_string_repr(stdout)
			), 
			command_str=command_str, full_command_str=full_command_str,
			stderr=stderr, stdout=stdout
		)

	@staticmethod
	def _get_command_str(cmd_str, args=None):
		if args is not None:
			cmd_str = windows_utils.format_command(cmd_str, **args)

		return cmd_str
