1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
|
# Copyright 2015 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.
"""This class defines the TestStationHost class."""
import logging
import os
import common
from autotest_lib.client.bin import local_host
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib.cros import retry
from autotest_lib.client.cros import constants as cros_constants
from autotest_lib.server.hosts import base_classes
from autotest_lib.server.hosts import moblab_host
from autotest_lib.server.hosts import ssh_host
# TODO(kevcheng): Update the creation method so it's not a research project
# determining the class inheritance model (same for factory.create_host).
def create_teststationhost(hostname, **kwargs):
"""Creates the TestStationHost object.
@param hostname: Hostname of the test station.
@param kwargs: Keyword args to pass to the testbed initialization.
@return: A Test Station Host object.
"""
classes = [TestStationHost]
if hostname == 'localhost':
classes.append(local_host.LocalHost)
else:
classes.append(ssh_host.SSHHost)
host_class = type('new_teststationhost', tuple(classes), {})
return host_class(hostname, **kwargs)
class TestStationHost(base_classes.Host):
"""This class represents a linux box accessible via ssh."""
def check_credentials(self, hostname):
"""Make sure teststation credentials work if we're doing ssh.
@param hostname: Hostname of the machine.
"""
if hostname != 'localhost':
try:
self.run('true')
except error.AutoservRunError:
# Some test stations may not have root access, try user adb.
logging.debug('Switching to user adb.')
self.user = 'adb'
def _initialize(self, hostname='localhost', *args, **dargs):
"""Initialize a Test Station Host.
This will create a Test Station Host. Hostname should always refer
to the host machine connected to the devices under test.
@param hostname: Hostname of the machine, default to localhost.
"""
logging.debug('Initializing Test Station Host running on host: %s.',
hostname)
# Do parent class initializations.
super(TestStationHost, self)._initialize(hostname=hostname, *args,
**dargs)
self.check_credentials(hostname)
# We'll want to do certain things differently if we're on a moblab.
self._is_host_moblab = None
# Keep track of whether the host was closed since multiple AdbHost
# might have an instance of this teststation.
self._is_closed = False
@property
def is_moblab(self):
"""Check if the host running adb command is a Moblab.
@return: True if the host running adb command is a Moblab, False
otherwise.
"""
if self._is_host_moblab is None:
try:
self.run('cat %s | grep -q moblab' % cros_constants.LSB_RELEASE)
self._is_host_moblab = True
except (error.AutoservRunError, error.AutotestHostRunError):
self._is_host_moblab = False
return self._is_host_moblab
def get_tmp_dir(self, parent='/var/tmp'):
"""Return pathname of a temporary directory on the test station.
If parent folder is supplied and the teststation is a moblab. Then
the parent will have the moblab tmp directory prepended to it.
@param parent: The parent dir to create the temporary dir.
@return: Path of the newly created temporary dir.
"""
if self.is_moblab:
parent = (moblab_host.MOBLAB_TMP_DIR if parent == '/tmp'
else os.path.join(moblab_host.MOBLAB_TMP_DIR,
parent.lstrip('/')))
return super(TestStationHost, self).get_tmp_dir(parent=parent)
def run(self, cmd, force_tty=True, *args, **dargs):
"""Run a command on the adb device.
This will run the command on the test station. This method only
exists to modify the command supplied if we're running a fastboot
command on a moblab, otherwise we leave the command untouched.
@param cmd: The command line string.
@param force_tty: Set to True to force pseudo-terminal allocation to
run the command. This allows the command running on remote host
to abort when the ssh command is timed out. Default is True.
@returns A CMDResult object or None if the call timed out and
ignore_timeout is True.
"""
# TODO (sbasi/kevcheng) - Make teststation_host check if running
# on Chrome OS, rather than MobLab when prepending sudo to fastboot.
if cmd.startswith('fastboot ') and self.is_moblab:
cmd = 'sudo -n ' + cmd
if force_tty:
dargs['options'] = dargs.get('options', '') + ' -t '
return super(TestStationHost, self).run(cmd, *args, **dargs)
@retry.retry(error.GenericHostRunError, timeout_min=10)
def download_file(self, src_url, dest_file, unzip=False, unzip_dest=None):
"""Download the given url.
@param src_url: The url to download from.
@param dest_file: Destination for the file to be downloaded to.
@param unzip: If True, unzip the downloaded file.
@param unzip_dest: Location to unzip the downloaded file to. If not
provided, dest_file's directory is used.
@returns: The path of the downloaded file on the teststation.
"""
try:
self.run('wget -q -O "%s" "%s"' % (dest_file, src_url))
readlink_result = self.run('readlink -f "%s"' % dest_file)
full_path = readlink_result.stdout.splitlines()[0]
if unzip:
unzip_dest = unzip_dest or os.path.dirname(full_path)
self.run('unzip "%s" -x -d "%s"' % (dest_file, unzip_dest))
return full_path
except:
# Delete the destination file if download failed.
self.run('rm -f "%s"' % dest_file)
raise
|
