blob: 5075cdf4a9aad1c47693acbe09fbc9caa166456d [file] [log] [blame]
# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Library containing functions to access a remote test device."""
import glob
import logging
import os
import shutil
import stat
import time
from chromite.lib import cros_build_lib
from chromite.lib import timeout_util
_path = os.path.dirname(os.path.realpath(__file__))
TEST_PRIVATE_KEY = os.path.normpath(
os.path.join(_path, '../ssh_keys/testing_rsa'))
del _path
REBOOT_MARKER = '/tmp/awaiting_reboot'
def CompileSSHConnectSettings(ConnectTimeout=30, ConnectionAttempts=4):
return ['-o', 'ConnectTimeout=%s' % ConnectTimeout,
'-o', 'ConnectionAttempts=%s' % ConnectionAttempts,
'-o', 'NumberOfPasswordPrompts=0',
'-o', 'Protocol=2',
'-o', 'ServerAliveInterval=10',
'-o', 'ServerAliveCountMax=3',
'-o', 'StrictHostKeyChecking=no',
'-o', 'UserKnownHostsFile=/dev/null', ]
class RemoteAccess(object):
"""Provides access to a remote test machine."""
def __init__(self, remote_host, tempdir, port=DEFAULT_SSH_PORT,
debug_level=logging.DEBUG, interactive=True):
"""Construct the object.
remote_host: The ip or hostname of the remote test machine. The test
machine should be running a ChromeOS test image.
tempdir: A directory that RemoteAccess can use to store temporary files.
It's the responsibility of the caller to remove it.
port: The ssh port of the test machine to connect to.
debug_level: Logging level to use for all RunCommand invocations.
interactive: If set to False, pass /dev/null into stdin for the sh cmd.
self.tempdir = tempdir
self.remote_host = remote_host
self.port = port
self.debug_level = debug_level
self.private_key = os.path.join(tempdir, os.path.basename(TEST_PRIVATE_KEY))
self.interactive = interactive
shutil.copyfile(TEST_PRIVATE_KEY, self.private_key)
os.chmod(self.private_key, stat.S_IRUSR)
def target_ssh_url(self):
return 'root@%s' % self.remote_host
def _GetSSHCmd(self, connect_settings=None):
if connect_settings is None:
connect_settings = CompileSSHConnectSettings()
cmd = (['ssh', '-p', str(self.port)] +
connect_settings +
['-i', self.private_key])
if not self.interactive:
return cmd
def RemoteSh(self, cmd, connect_settings=None, error_code_ok=False,
ssh_error_ok=False, debug_level=None):
"""Run a sh command on the remote device through ssh.
cmd: The command string or list to run.
connect_settings: The SSH connect settings to use.
error_code_ok: Does not throw an exception when the command exits with a
non-zero returncode. This does not cover the case where
the ssh command itself fails (return code 255).
See ssh_error_ok.
ssh_error_ok: Does not throw an exception when the ssh command itself
fails (return code 255).
debug_level: See cros_build_lib.RunCommand documentation.
A CommandResult object. The returncode is the returncode of the command,
or 255 if ssh encountered an error (could not connect, connection
interrupted, etc.)
RunCommandError when error is not ignored through error_code_ok and
ssh_error_ok flags.
if not debug_level:
debug_level = self.debug_level
ssh_cmd = self._GetSSHCmd(connect_settings)
if isinstance(cmd, basestring):
ssh_cmd += [self.target_ssh_url, '--', cmd]
ssh_cmd += [self.target_ssh_url, '--'] + cmd
result = cros_build_lib.RunCommandCaptureOutput(
ssh_cmd, debug_level=debug_level)
except cros_build_lib.RunCommandError as e:
if ((e.result.returncode == SSH_ERROR_CODE and ssh_error_ok) or
(e.result.returncode and e.result.returncode != SSH_ERROR_CODE
and error_code_ok)):
result = e.result
return result
def LearnBoard(self):
"""Grab the board reported by the remote device.
in the case of multiple matches, uses the first one. In the case of no
entry, returns an empty string.
result = self.RemoteSh('grep CHROMEOS_RELEASE_BOARD /etc/lsb-release')
# In the case of multiple matches, use the first one.
output = result.output.splitlines()
if len(output) > 1:
logging.debug('More than one board entry found! Using the first one.')
return output[0].strip().partition('=')[-1]
def _CheckIfRebooted(self):
""""Checks whether a remote device has rebooted successfully.
This uses a rapidly-retried SSH connection, which will wait for at most
about ten seconds. If the network returns an error (e.g. host unreachable)
the actual delay may be shorter.
Whether the device has successfully rebooted.
# In tests SSH seems to be waiting rather longer than would be expected
# from these parameters. These values produce a ~5 second wait.
connect_settings = CompileSSHConnectSettings(
cmd = "[ ! -e '%s' ]" % REBOOT_MARKER
result = self.RemoteSh(cmd, connect_settings=connect_settings,
error_code_ok=True, ssh_error_ok=True)
errors = {0: 'Reboot complete.',
1: 'Device has not yet shutdown.',
255: 'Cannot connect to device; reboot in progress.'}
if result.returncode not in errors:
raise Exception('Unknown error code %s returned by %s.'
% (result.returncode, cmd))[result.returncode])
return result.returncode == 0
def RemoteReboot(self):
"""Reboot the remote device."""'Rebooting %s...', self.remote_host)
self.RemoteSh('touch %s && reboot' % REBOOT_MARKER)
timeout_util.WaitForReturnTrue(self._CheckIfRebooted, REBOOT_MAX_WAIT,
except timeout_util.TimeoutError:
cros_build_lib.Die('Reboot has not completed after %s seconds; giving up.'
def Rsync(self, src, dest, to_local=False, follow_symlinks=False,
inplace=False, verbose=False, sudo=False, **kwargs):
"""Rsync a path to the remote device.
Rsync a path to the remote device. If |to_local| is set True, it
rsyncs the path from the remote device to the local machine.
src: The local src directory.
dest: The remote dest directory.
to_local: If set, rsync remote path to local path.
follow_symlinks: If set, transform symlinks into referent
path. Otherwise, copy symlinks as symlinks.
inplace: If set, cause rsync to overwrite the dest files in place. This
conserves space, but has some side effects - see rsync man page.
verbose: If set, print more verbose output during rsync file transfer.
sudo: If set, invoke the command via sudo.
**kwargs: See cros_build_lib.RunCommand documentation.
kwargs.setdefault('debug_level', self.debug_level)
ssh_cmd = ' '.join(self._GetSSHCmd())
rsync_cmd = ['rsync', '--recursive', '--perms', '--verbose',
'--times', '--compress', '--omit-dir-times',
'--exclude', '.svn']
rsync_cmd.append('--copy-links' if follow_symlinks else '--links')
# In cases where the developer sets up a ssh daemon manually on a device
# with a dev image, the ssh login $PATH can be incorrect, and the target
# rsync will not be found. So we try to provide the right $PATH here.
rsync_cmd += ['--rsync-path', 'PATH=/usr/local/bin:$PATH rsync']
if verbose:
if inplace:
if to_local:
rsync_cmd += ['--rsh', ssh_cmd,
'[%s]:%s' % (self.target_ssh_url, src), dest]
rsync_cmd += ['--rsh', ssh_cmd, src,
'[%s]:%s' % (self.target_ssh_url, dest)]
rc_func = cros_build_lib.RunCommand
if sudo:
rc_func = cros_build_lib.SudoRunCommand
return rc_func(rsync_cmd, print_cmd=verbose, **kwargs)
def RsyncToLocal(self, *args, **kwargs):
"""Rsync a path from the remote device to the local machine."""
return self.Rsync(*args, to_local=kwargs.pop('to_local', True), **kwargs)
def Scp(self, src, dest, recursive=False, verbose=False, debug_level=None,
"""Scp a file or directory to the remote device.
src: The local src file or directory.
dest: The remote dest location.
recursive: Whether to recursively copy entire directories.
verbose: If set, print more verbose output during scp file transfer.
debug_level: See cros_build_lib.RunCommand documentation.
sudo: If set, invoke the command via sudo.
A CommandResult object containing the information and return code of
the scp command.
if not debug_level:
debug_level = self.debug_level
scp_cmd = ['scp', '-p'] + CompileSSHConnectSettings(ConnectTimeout=60)
if recursive:
if verbose:
scp_cmd += glob.glob(src) + ['%s:%s' % (self.target_ssh_url, dest)]
rc_func = cros_build_lib.RunCommand
if sudo:
rc_func = cros_build_lib.SudoRunCommand
return rc_func(scp_cmd, debug_level=debug_level, print_cmd=verbose)