xref: /aosp_15_r20/external/cronet/build/fuchsia/test/flash_device.py (revision 6777b5387eb2ff775bb5750e3f5d96f37fb7352b)

#!/usr/bin/env vpython3
# Copyright 2022 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Implements commands for flashing a Fuchsia device."""

import argparse
import logging
import os
import subprocess
import sys

from typing import Optional, Tuple

import common
from boot_device import BootMode, StateTransitionError, boot_device
from common import get_system_info, find_image_in_sdk, \
                   register_device_args
from compatible_utils import get_sdk_hash, running_unattended
from lockfile import lock

# Flash-file lock. Used to restrict number of flash operations per host.
# File lock should be marked as stale after 15 mins.
_FF_LOCK = os.path.join('/tmp', 'flash.lock')
_FF_LOCK_STALE_SECS = 60 * 15
_FF_LOCK_ACQ_TIMEOUT = _FF_LOCK_STALE_SECS


def _get_system_info(target: Optional[str],
                     serial_num: Optional[str]) -> Tuple[str, str]:
    """Retrieves installed OS version from device.

    Args:
        target: Target to get system info of.
        serial_num: Serial number of device to get system info of.
    Returns:
        Tuple of strings, containing (product, version number).
    """

    if running_unattended():
        try:
            boot_device(target, BootMode.REGULAR, serial_num)
        except (subprocess.CalledProcessError, StateTransitionError):
            logging.warning('Could not boot device. Assuming in fastboot')
            return ('', '')

    return get_system_info(target)


def _update_required(
        os_check,
        system_image_dir: Optional[str],
        target: Optional[str],
        serial_num: Optional[str] = None) -> Tuple[bool, Optional[str]]:
    """Returns True if a system update is required and path to image dir."""

    if os_check == 'ignore':
        return False, system_image_dir
    if not system_image_dir:
        raise ValueError('System image directory must be specified.')
    if not os.path.exists(system_image_dir):
        logging.warning(
            'System image directory does not exist. Assuming it\'s '
            'a product-bundle name and dynamically searching for '
            'image directory')
        path = find_image_in_sdk(system_image_dir)
        if not path:
            raise FileNotFoundError(
                f'System image directory {system_image_dir} could not'
                'be found')
        system_image_dir = path
    if (os_check == 'check'
            and get_sdk_hash(system_image_dir) == _get_system_info(
                target, serial_num)):
        return False, system_image_dir
    return True, system_image_dir


def _run_flash_command(system_image_dir: str, target_id: Optional[str]):
    """Helper function for running `ffx target flash`."""
    logging.info('Flashing %s to %s', system_image_dir, target_id)
    # Flash only with a file lock acquired.
    # This prevents multiple fastboot binaries from flashing concurrently,
    # which should increase the odds of flashing success.
    with lock(_FF_LOCK, timeout=_FF_LOCK_ACQ_TIMEOUT):
        # The ffx.fastboot.inline_target has negative impact when ffx
        # discovering devices in fastboot, so it's inlined here to limit its
        # scope. See the discussion in https://fxbug.dev/issues/317228141.
        logging.info(
            'Flash result %s',
            common.run_ffx_command(cmd=('target', 'flash', '-b',
                                        system_image_dir,
                                        '--no-bootloader-reboot'),
                                   target_id=target_id,
                                   configs=['ffx.fastboot.inline_target=true'],
                                   capture_output=True).stdout)


def update(system_image_dir: str,
           os_check: str,
           target: Optional[str],
           serial_num: Optional[str] = None) -> None:
    """Conditionally updates target given.

    Args:
        system_image_dir: string, path to image directory.
        os_check: <check|ignore|update>, which decides how to update the device.
        target: Node-name string indicating device that should be updated.
        serial_num: String of serial number of device that should be updated.
    """
    needs_update, actual_image_dir = _update_required(os_check,
                                                      system_image_dir, target,
                                                      serial_num)
    logging.info('update_required %s, actual_image_dir %s', needs_update,
                 actual_image_dir)
    if not needs_update:
        return
    if serial_num:
        boot_device(target, BootMode.BOOTLOADER, serial_num)
        _run_flash_command(system_image_dir, serial_num)
    else:
        _run_flash_command(system_image_dir, target)


def register_update_args(arg_parser: argparse.ArgumentParser,
                         default_os_check: Optional[str] = 'check') -> None:
    """Register common arguments for device updating."""
    serve_args = arg_parser.add_argument_group('update',
                                               'device updating arguments')
    serve_args.add_argument('--system-image-dir',
                            help='Specify the directory that contains the '
                            'Fuchsia image used to flash the device. Only '
                            'needs to be specified if "os_check" is not '
                            '"ignore".')
    serve_args.add_argument('--serial-num',
                            default=os.environ.get('FUCHSIA_FASTBOOT_SERNUM'),
                            help='Serial number of the device. Should be '
                            'specified for devices that do not have an image '
                            'flashed.')
    serve_args.add_argument('--os-check',
                            choices=['check', 'update', 'ignore'],
                            default=default_os_check,
                            help='Sets the OS version enforcement policy. If '
                            '"check", then the deployment process will halt '
                            'if the target\'s version does not match. If '
                            '"update", then the target device will '
                            'be reflashed. If "ignore", then the OS version '
                            'will not be checked.')


def main():
    """Stand-alone function for flashing a device."""
    parser = argparse.ArgumentParser()
    register_device_args(parser)
    register_update_args(parser, default_os_check='update')
    args = parser.parse_args()
    update(args.system_image_dir, args.os_check, args.target_id,
           args.serial_num)


if __name__ == '__main__':
    sys.exit(main())