xref: /aosp_15_r20/external/autotest/server/hosts/android_host.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# Lint as: python2, python3
# Copyright (c) 2022 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.
#
# Expects to be run in an environment with sudo and no interactive password
# prompt, such as within the Chromium OS development chroot.

import logging
import socket

import common

from autotest_lib.client.common_lib import error
from autotest_lib.server.hosts import host_info
from autotest_lib.server.hosts import attached_device_host
from autotest_lib.server.hosts import android_constants
from autotest_lib.server.hosts import base_classes


class AndroidHost(base_classes.Host):
    """Host class for Android devices"""
    PHONE_STATION_LABEL_PREFIX = "associated_hostname"
    SERIAL_NUMBER_LABEL_PREFIX = "serial_number"
    # adb auth key path on the phone_station.
    ADB_KEY_PATH = '/var/lib/android_keys'

    def __init__(self,
                 hostname,
                 host_info_store=None,
                 android_args=None,
                 *args,
                 **dargs):
        """Construct a AndroidHost object.

        Args:
            hostname: Hostname of the Android phone.
            host_info_store: Optional host_info.CachingHostInfoStore object
                             to obtain / update host information.
            android_args: Android args for local test run.
        """
        self.hostname = hostname
        super(AndroidHost, self).__init__(*args, **dargs)
        self.host_info_store = (host_info_store
                                or host_info.InMemoryHostInfoStore())
        self.associated_hostname = None
        self.serial_number = None
        self.phone_station_ssh_port = None
        # For local test, android_args are passed in.
        if android_args:
            self._read_essential_data_from_args_dict(android_args)
        else:
            self._read_essential_data_from_host_info_store()
        # Since we won't be ssh into an Android device directly, all the
        # communication will be handled by run ADB CLI on the phone
        # station(chromebox or linux machine) that physically connected
        # to the Android devices via USB cable. So we need to setup an
        # AttachedDeviceHost for phone station as ssh proxy.
        self.phone_station = self._create_phone_station_host_proxy()
        self.adb_tcp_mode = False
        self.usb_dev_path = None
        self.closed = False

    def _create_phone_station_host_proxy(self):
        logging.info('Creating host for phone station %s',
                     self.associated_hostname)
        return attached_device_host.AttachedDeviceHost(
                hostname=self.associated_hostname,
                serial_number=self.serial_number,
                phone_station_ssh_port=self.phone_station_ssh_port)

    def _read_essential_data_from_args_dict(self, android_args):
        self.associated_hostname = android_args.get(
                android_constants.ANDROID_PHONE_STATION_ATTR)
        self.phone_station_ssh_port = android_args.get(
                android_constants.ANDROID_PHONE_STATION_SSH_PORT_ATTR)
        self.serial_number = android_args.get(
                android_constants.ANDROID_SERIAL_NUMBER_ATTR)

    def _read_essential_data_from_host_info_store(self):
        info = self.host_info_store.get()
        self.associated_hostname = info.get_label_value(
                self.PHONE_STATION_LABEL_PREFIX)
        if not self.associated_hostname:
            raise error.AutoservError(
                    'Failed to initialize Android host due to'
                    ' associated_hostname is not found in host_info_store.')
        self.serial_number = info.get_label_value(
                self.SERIAL_NUMBER_LABEL_PREFIX)
        if not self.serial_number:
            raise error.AutoservError(
                    'Failed to initialize Android host due to'
                    ' serial_number is not found in host_info_store.')

    def adb_over_tcp(self, port=5555, persist_reboot=False):
        """Restart adb server listening on a TCP port.

        Args:
            port: Tcp port for adb server to listening on, default value
                  is 5555 which is the default TCP/IP port for adb.
            persist_reboot: True for adb over tcp to continue listening
                            after the device reboots.
        """
        port = str(port)
        if persist_reboot:
            self.run_adb_command('shell setprop persist.adb.tcp.port %s' %
                                 port)
            self.run_adb_command('shell setprop ctl.restart adbd')
            self.wait_for_transport_state()

        self.run_adb_command('tcpip %s' % port)
        self.adb_tcp_mode = True

    def cache_usb_dev_path(self):
        """
        Read and cache usb devpath for the Android device.
        """
        cmd = 'adb devices -l | grep %s' % self.serial_number
        res = self.phone_station.run(cmd)
        for line in res.stdout.strip().split('\n'):
            if len(line.split()) > 2 and line.split()[1] == 'device':
                self.usb_dev_path = line.split()[2]
                logging.info('USB devpath: %s', self.usb_dev_path)
                break
        if not self.usb_dev_path:
            logging.warning(
                    'Failed to collect usbdev path of the Android device.')

    def ensure_device_connectivity(self):
        """Ensure we can interact with the Android device via adb and
        the device is in the expected state.
        """
        res = self.run_adb_command('get-state')
        state = res.stdout.strip()
        logging.info('Android device state from adb: %s', state)
        return state == 'device'

    def get_wifi_ip_address(self):
        """Get ipv4 address from the Android device"""
        res = self.run_adb_command('shell ip route')
        # An example response would looks like: "192.168.86.0/24 dev wlan0"
        # " proto kernel scope link src 192.168.86.22 \n"
        ip_string = res.stdout.strip().split(' ')[-1]
        logging.info('Ip address collected from the Android device: %s',
                     ip_string)
        try:
            socket.inet_aton(ip_string)
        except (OSError, ValueError, socket.error):
            raise error.AutoservError(
                    'Failed to get ip address from the Android device.')
        return ip_string

    def job_start(self):
        """This method is called from create_host factory when
        construct the host object. We need to override it since actions
        like copy /var/log/messages are not applicable on Android devices.
        """
        logging.info('Skip standard job_start actions for Android host.')

    def restart_adb_server(self):
        """Restart adb server from the phone station"""
        self.stop_adb_server()
        self.start_adb_server()

    def run_adb_command(self, adb_command, ignore_status=False):
        """Run adb command on the Android device.

        Args:
            adb_command: adb commands to execute on the Android device.

        Returns:
            An autotest_lib.client.common_lib.utils.CmdResult object.
        """
        # When use adb to interact with an Android device, we prefer to use
        # devpath to distinguish the particular device as the serial number
        # is not guaranteed to be unique.
        if self.usb_dev_path:
            command = 'adb -s %s %s' % (self.usb_dev_path, adb_command)
        else:
            command = 'adb -s %s %s' % (self.serial_number, adb_command)
        return self.phone_station.run(command, ignore_status=ignore_status)

    def wait_for_transport_state(self, transport='usb', state='device'):
        """
        Wait for a device to reach a desired state.

        Args:
            transport: usb, local, any
            state: device, recovery, sideload, bootloader

        """
        self.run_adb_command('wait-for-%s-%s' % (transport, state))

    def start_adb_server(self):
        """Start adb server from the phone station."""
        # Adb home is created upon CrOS login, however on labstation we
        # never login so we'll need to ensure the adb home is exist before
        # starting adb server.
        self.phone_station.run("mkdir -p /run/arc/adb")
        self.phone_station.run("ADB_VENDOR_KEYS=%s adb start-server" %
                               self.ADB_KEY_PATH)
        # Logging states of all attached devices.
        self.phone_station.run('adb devices')

    def stop_adb_server(self):
        """Stop adb server from the phone station."""
        self.phone_station.run("adb kill-server")

    def setup_for_cross_device_tests(self, adb_persist_reboot=False):
        """
        Setup the Android phone for Cross Device tests.

        Ensures the phone can connect to its labstation and sets up
        adb-over-tcp.

        Returns:
            IP Address of Phone.
        """
        dut_out = self.phone_station.run('echo True').stdout.strip()
        if dut_out != 'True':
            raise error.TestError('phone station stdout != True (got: %s)',
                                  dut_out)

        self.restart_adb_server()
        self.cache_usb_dev_path()
        self.ensure_device_connectivity()
        ip_address = self.get_wifi_ip_address()
        self.adb_over_tcp(persist_reboot=adb_persist_reboot)
        return ip_address

    def close(self):
        """Clean up Android host and its phone station proxy host."""
        if self.closed:
            logging.debug('Android host %s already closed.', self.hostname)
            return
        try:
            if self.adb_tcp_mode:
                # In some rare cases, leave the Android device in adb over tcp
                # mode may break USB connection so we want to always reset adb
                # to usb mode before teardown.
                self.run_adb_command('usb', ignore_status=True)
            self.stop_adb_server()
            if self.phone_station:
                self.phone_station.close()
            self.closed = True
        finally:
            super(AndroidHost, self).close()

    @staticmethod
    def get_android_arguments(args_dict):
        """Extract android args from `args_dict` and return the result.

        Recommended usage in control file:
            args_dict = utils.args_to_dict(args)
            android_args = hosts.Android.get_android_arguments(args_dict)
            host = hosts.create_host(machine, android_args=android_args)

        Args:
            args_dict: A dict of test args.

        Returns:
            An dict of android related args.
        """
        android_args = {
                key: args_dict[key]
                for key in android_constants.ALL_ANDROID_ATTRS
                if key in args_dict
        }
        for attr in android_constants.CRITICAL_ANDROID_ATTRS:
            if attr not in android_args or not android_args.get(attr):
                raise error.AutoservError("Critical attribute %s is missing"
                                          " from android_args." % attr)
        return android_args