1# Lint as: python2, python3 2# Copyright (c) 2022 The Chromium OS Authors. All rights reserved. 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5# 6# Expects to be run in an environment with sudo and no interactive password 7# prompt, such as within the Chromium OS development chroot. 8 9import logging 10import socket 11 12import common 13 14from autotest_lib.client.common_lib import error 15from autotest_lib.server.hosts import host_info 16from autotest_lib.server.hosts import attached_device_host 17from autotest_lib.server.hosts import android_constants 18from autotest_lib.server.hosts import base_classes 19 20 21class AndroidHost(base_classes.Host): 22 """Host class for Android devices""" 23 PHONE_STATION_LABEL_PREFIX = "associated_hostname" 24 SERIAL_NUMBER_LABEL_PREFIX = "serial_number" 25 # adb auth key path on the phone_station. 26 ADB_KEY_PATH = '/var/lib/android_keys' 27 28 def __init__(self, 29 hostname, 30 host_info_store=None, 31 android_args=None, 32 *args, 33 **dargs): 34 """Construct a AndroidHost object. 35 36 Args: 37 hostname: Hostname of the Android phone. 38 host_info_store: Optional host_info.CachingHostInfoStore object 39 to obtain / update host information. 40 android_args: Android args for local test run. 41 """ 42 self.hostname = hostname 43 super(AndroidHost, self).__init__(*args, **dargs) 44 self.host_info_store = (host_info_store 45 or host_info.InMemoryHostInfoStore()) 46 self.associated_hostname = None 47 self.serial_number = None 48 self.phone_station_ssh_port = None 49 # For local test, android_args are passed in. 50 if android_args: 51 self._read_essential_data_from_args_dict(android_args) 52 else: 53 self._read_essential_data_from_host_info_store() 54 # Since we won't be ssh into an Android device directly, all the 55 # communication will be handled by run ADB CLI on the phone 56 # station(chromebox or linux machine) that physically connected 57 # to the Android devices via USB cable. So we need to setup an 58 # AttachedDeviceHost for phone station as ssh proxy. 59 self.phone_station = self._create_phone_station_host_proxy() 60 self.adb_tcp_mode = False 61 self.usb_dev_path = None 62 self.closed = False 63 64 def _create_phone_station_host_proxy(self): 65 logging.info('Creating host for phone station %s', 66 self.associated_hostname) 67 return attached_device_host.AttachedDeviceHost( 68 hostname=self.associated_hostname, 69 serial_number=self.serial_number, 70 phone_station_ssh_port=self.phone_station_ssh_port) 71 72 def _read_essential_data_from_args_dict(self, android_args): 73 self.associated_hostname = android_args.get( 74 android_constants.ANDROID_PHONE_STATION_ATTR) 75 self.phone_station_ssh_port = android_args.get( 76 android_constants.ANDROID_PHONE_STATION_SSH_PORT_ATTR) 77 self.serial_number = android_args.get( 78 android_constants.ANDROID_SERIAL_NUMBER_ATTR) 79 80 def _read_essential_data_from_host_info_store(self): 81 info = self.host_info_store.get() 82 self.associated_hostname = info.get_label_value( 83 self.PHONE_STATION_LABEL_PREFIX) 84 if not self.associated_hostname: 85 raise error.AutoservError( 86 'Failed to initialize Android host due to' 87 ' associated_hostname is not found in host_info_store.') 88 self.serial_number = info.get_label_value( 89 self.SERIAL_NUMBER_LABEL_PREFIX) 90 if not self.serial_number: 91 raise error.AutoservError( 92 'Failed to initialize Android host due to' 93 ' serial_number is not found in host_info_store.') 94 95 def adb_over_tcp(self, port=5555, persist_reboot=False): 96 """Restart adb server listening on a TCP port. 97 98 Args: 99 port: Tcp port for adb server to listening on, default value 100 is 5555 which is the default TCP/IP port for adb. 101 persist_reboot: True for adb over tcp to continue listening 102 after the device reboots. 103 """ 104 port = str(port) 105 if persist_reboot: 106 self.run_adb_command('shell setprop persist.adb.tcp.port %s' % 107 port) 108 self.run_adb_command('shell setprop ctl.restart adbd') 109 self.wait_for_transport_state() 110 111 self.run_adb_command('tcpip %s' % port) 112 self.adb_tcp_mode = True 113 114 def cache_usb_dev_path(self): 115 """ 116 Read and cache usb devpath for the Android device. 117 """ 118 cmd = 'adb devices -l | grep %s' % self.serial_number 119 res = self.phone_station.run(cmd) 120 for line in res.stdout.strip().split('\n'): 121 if len(line.split()) > 2 and line.split()[1] == 'device': 122 self.usb_dev_path = line.split()[2] 123 logging.info('USB devpath: %s', self.usb_dev_path) 124 break 125 if not self.usb_dev_path: 126 logging.warning( 127 'Failed to collect usbdev path of the Android device.') 128 129 def ensure_device_connectivity(self): 130 """Ensure we can interact with the Android device via adb and 131 the device is in the expected state. 132 """ 133 res = self.run_adb_command('get-state') 134 state = res.stdout.strip() 135 logging.info('Android device state from adb: %s', state) 136 return state == 'device' 137 138 def get_wifi_ip_address(self): 139 """Get ipv4 address from the Android device""" 140 res = self.run_adb_command('shell ip route') 141 # An example response would looks like: "192.168.86.0/24 dev wlan0" 142 # " proto kernel scope link src 192.168.86.22 \n" 143 ip_string = res.stdout.strip().split(' ')[-1] 144 logging.info('Ip address collected from the Android device: %s', 145 ip_string) 146 try: 147 socket.inet_aton(ip_string) 148 except (OSError, ValueError, socket.error): 149 raise error.AutoservError( 150 'Failed to get ip address from the Android device.') 151 return ip_string 152 153 def job_start(self): 154 """This method is called from create_host factory when 155 construct the host object. We need to override it since actions 156 like copy /var/log/messages are not applicable on Android devices. 157 """ 158 logging.info('Skip standard job_start actions for Android host.') 159 160 def restart_adb_server(self): 161 """Restart adb server from the phone station""" 162 self.stop_adb_server() 163 self.start_adb_server() 164 165 def run_adb_command(self, adb_command, ignore_status=False): 166 """Run adb command on the Android device. 167 168 Args: 169 adb_command: adb commands to execute on the Android device. 170 171 Returns: 172 An autotest_lib.client.common_lib.utils.CmdResult object. 173 """ 174 # When use adb to interact with an Android device, we prefer to use 175 # devpath to distinguish the particular device as the serial number 176 # is not guaranteed to be unique. 177 if self.usb_dev_path: 178 command = 'adb -s %s %s' % (self.usb_dev_path, adb_command) 179 else: 180 command = 'adb -s %s %s' % (self.serial_number, adb_command) 181 return self.phone_station.run(command, ignore_status=ignore_status) 182 183 def wait_for_transport_state(self, transport='usb', state='device'): 184 """ 185 Wait for a device to reach a desired state. 186 187 Args: 188 transport: usb, local, any 189 state: device, recovery, sideload, bootloader 190 191 """ 192 self.run_adb_command('wait-for-%s-%s' % (transport, state)) 193 194 def start_adb_server(self): 195 """Start adb server from the phone station.""" 196 # Adb home is created upon CrOS login, however on labstation we 197 # never login so we'll need to ensure the adb home is exist before 198 # starting adb server. 199 self.phone_station.run("mkdir -p /run/arc/adb") 200 self.phone_station.run("ADB_VENDOR_KEYS=%s adb start-server" % 201 self.ADB_KEY_PATH) 202 # Logging states of all attached devices. 203 self.phone_station.run('adb devices') 204 205 def stop_adb_server(self): 206 """Stop adb server from the phone station.""" 207 self.phone_station.run("adb kill-server") 208 209 def setup_for_cross_device_tests(self, adb_persist_reboot=False): 210 """ 211 Setup the Android phone for Cross Device tests. 212 213 Ensures the phone can connect to its labstation and sets up 214 adb-over-tcp. 215 216 Returns: 217 IP Address of Phone. 218 """ 219 dut_out = self.phone_station.run('echo True').stdout.strip() 220 if dut_out != 'True': 221 raise error.TestError('phone station stdout != True (got: %s)', 222 dut_out) 223 224 self.restart_adb_server() 225 self.cache_usb_dev_path() 226 self.ensure_device_connectivity() 227 ip_address = self.get_wifi_ip_address() 228 self.adb_over_tcp(persist_reboot=adb_persist_reboot) 229 return ip_address 230 231 def close(self): 232 """Clean up Android host and its phone station proxy host.""" 233 if self.closed: 234 logging.debug('Android host %s already closed.', self.hostname) 235 return 236 try: 237 if self.adb_tcp_mode: 238 # In some rare cases, leave the Android device in adb over tcp 239 # mode may break USB connection so we want to always reset adb 240 # to usb mode before teardown. 241 self.run_adb_command('usb', ignore_status=True) 242 self.stop_adb_server() 243 if self.phone_station: 244 self.phone_station.close() 245 self.closed = True 246 finally: 247 super(AndroidHost, self).close() 248 249 @staticmethod 250 def get_android_arguments(args_dict): 251 """Extract android args from `args_dict` and return the result. 252 253 Recommended usage in control file: 254 args_dict = utils.args_to_dict(args) 255 android_args = hosts.Android.get_android_arguments(args_dict) 256 host = hosts.create_host(machine, android_args=android_args) 257 258 Args: 259 args_dict: A dict of test args. 260 261 Returns: 262 An dict of android related args. 263 """ 264 android_args = { 265 key: args_dict[key] 266 for key in android_constants.ALL_ANDROID_ATTRS 267 if key in args_dict 268 } 269 for attr in android_constants.CRITICAL_ANDROID_ATTRS: 270 if attr not in android_args or not android_args.get(attr): 271 raise error.AutoservError("Critical attribute %s is missing" 272 " from android_args." % attr) 273 return android_args 274