Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:param transfer_timeout: The time duration (in seconds) to wait for the
transfer to complete.
:param login_timeout: The maximal time duration (in seconds) to wait for
each step of the login procedure (i.e. the "Are you sure" prompt
or the password prompt)
:raise: Whatever _remote_scp() raises
"""
logging.debug("Trying to SCP with command '%s', timeout %ss",
command, transfer_timeout)
if log_filename:
output_func = utils_misc.log_line
output_params = (log_filename,)
else:
output_func = None
output_params = ()
session = aexpect.Expect(command,
output_func=output_func,
output_params=output_params)
try:
_remote_scp(session, password_list, transfer_timeout, login_timeout)
finally:
session.close()
@param transfer_timeout: The time duration (in seconds) to wait for the
transfer to complete.
@param login_timeout: The maximal time duration (in seconds) to wait for
each step of the login procedure (i.e. the "Are you sure" prompt
or the password prompt)
@raise: Whatever _remote_scp() raises
"""
logging.debug("Trying to SCP with command '%s', timeout %ss",
command, transfer_timeout)
if log_filename:
output_func = utils_misc.log_line
output_params = (log_filename,)
else:
output_func = None
output_params = ()
session = aexpect.Expect(command,
output_func=output_func,
output_params=output_params)
try:
_remote_scp(session, password_list, transfer_timeout, login_timeout)
finally:
session.close()
def test_kill_stopped_sleep(self):
proc = aexpect.Expect("%s run 60 --job-results-dir %s "
"--external-runner %s --sysinfo=off "
"--job-timeout 3"
% (AVOCADO, self.tmpdir.name, SLEEP_BINARY))
proc.read_until_output_matches([r"\(1/1\)"], timeout=3,
internal_timeout=0.01)
# We need pid of the avocado process, not the shell executing it
avocado_shell = psutil.Process(proc.get_pid())
avocado_proc = avocado_shell.children()[0]
pid = avocado_proc.pid
os.kill(pid, signal.SIGTSTP) # This freezes the process
# The deadline is 3s timeout + 10s test postprocess before kill +
# 10s reserve for additional steps (still below 60s)
deadline = time.time() + 20
while time.time() < deadline:
if not proc.is_alive():
break
:param transfer_timeout: The time duration (in seconds) to wait for the
transfer to complete.
:param login_timeout: The maximal time duration (in seconds) to wait for
each step of the login procedure (i.e. the "Are you sure" prompt
or the password prompt)
:raise: Whatever _remote_scp() raises
"""
logging.debug("Trying to SCP with command '%s', timeout %ss",
command, transfer_timeout)
if log_filename:
output_func = utils_misc.log_line
output_params = (log_filename,)
else:
output_func = None
output_params = ()
session = aexpect.Expect(command,
output_func=output_func,
output_params=output_params)
try:
_remote_scp(session, password_list, transfer_timeout, login_timeout)
finally:
session.close()
@param transfer_timeout: The time duration (in seconds) to wait for the
transfer to complete.
@param login_timeout: The maximal time duration (in seconds) to wait for
each step of the login procedure (i.e. the "Are you sure" prompt
or the password prompt)
@raise: Whatever _remote_scp() raises
"""
logging.debug("Trying to SCP with command '%s', timeout %ss",
command, transfer_timeout)
if log_filename:
output_func = log_line
output_params = (log_filename,)
else:
output_func = None
output_params = ()
session = aexpect.Expect(command,
output_func=output_func,
output_params=output_params)
try:
_remote_scp(session, password_list, transfer_timeout, login_timeout)
finally:
session.close()
@param transfer_timeout: The time duration (in seconds) to wait for the
transfer to complete.
@param login_timeout: The maximal time duration (in seconds) to wait for
each step of the login procedure (i.e. the "Are you sure" prompt
or the password prompt)
@raise: Whatever _remote_scp() raises
"""
logging.debug("Trying to SCP with command '%s', timeout %ss",
command, transfer_timeout)
if log_filename:
output_func = utils_misc.log_line
output_params = (log_filename,)
else:
output_func = None
output_params = ()
session = aexpect.Expect(command,
output_func=output_func,
output_params=output_params)
try:
_remote_scp(session, password_list, transfer_timeout, login_timeout)
finally:
session.close()
:param transfer_timeout: The time duration (in seconds) to wait for the
transfer to complete.
:param login_timeout: The maximal time duration (in seconds) to wait for
each step of the login procedure (i.e. the "Are you sure" prompt
or the password prompt)
:raise: Whatever _remote_scp() raises
"""
logging.debug("Trying to SCP with command '%s', timeout %ss",
command, transfer_timeout)
if log_filename:
output_func = utils.log_line
output_params = (log_filename,)
else:
output_func = None
output_params = ()
session = aexpect.Expect(command,
output_func=output_func,
output_params=output_params)
try:
_remote_scp(session, password_list, transfer_timeout, login_timeout)
finally:
session.close()
if path is None:
path = data_dir.get_tmp_dir()
if client == "ssh":
cmd = ("ssh -o UserKnownHostsFile=/dev/null "
"-o PreferredAuthentications=password "
"-p %s %s@%s %s agent_base64" %
(port, username, host, os.path.join(path, "remote_runner.py")))
elif client == "telnet":
cmd = "telnet -l %s %s %s" % (username, host, port)
elif client == "nc":
cmd = "nc %s %s" % (host, port)
else:
raise LoginBadClientError(client)
logging.debug("Login command: '%s'", cmd)
session = aexpect.Expect(cmd, linesep=linesep)
try:
handle_prompts(session, username, password, prompt, timeout)
except Exception:
session.close()
raise
if log_filename:
log_file = utils_misc.get_log_filename(log_filename)
session.set_output_func(utils_misc.log_line)
session.set_output_params((log_file,))
session.set_log_file(os.path.basename(log_file))
session.send_ctrl("raw")
# Wrap io interfaces.
inw = messenger.StdIOWrapperInBase64(session._get_fd("tail"))
outw = AexpectIOWrapperOut(session)
# Create commander