xref: /external/autotest/client/site_tests/usbpd_GFU/usbpd_GFU.py

# Copyright (c) 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.

import collections
import glob
import logging
import re
import time

from autotest_lib.client.bin import test
from autotest_lib.client.common_lib import error, utils
from autotest_lib.client.cros import ec as cros_ec, cros_logging


class usbpd_GFU(test.test):
    """Integration test for USB-PD Google Firmware Update (GFU).

    Test should:
    - interrogate what firmware's are available for each device and for each:
      1. Use ectool's flashpd to write RW with that to mimic old hw
         - Validate that kernel driver successfully updates to latest RW.
      2. Erase RW and see update as well.

    TODO:
      3. Check that update is checked after S2R.
    """

    version = 1

    FW_PATH = '/lib/firmware/cros-pd'
    # <device>_v<major>.<minor>.<build>-<commit SHA>
    FW_NAME_RE = r'%s/(\w+)_v(\d+)\.(\d+)\.(\d+)-([0-9a-f]+).*' % (FW_PATH)
    GOOGLE_VID = '0x18d1'
    MAX_UPDATE_SECS = 80
    FW_UP_DNAME = 'cros_ec_pd_update'
    # TODO(tbroch) This will be change once cros_ec_pd_update is abstracted from
    # ACPI driver.  Will need to fix this once it happens.
    FW_UP_DISABLE_PATH = '/sys/devices/LNXSYSTM:00/device:00/PNP0A08:00/device:1e/PNP0C09:00/GOOG0003:00/disable'

    # TODO(tbroch) find better way to build this or we'll have to edit test for
    # each new PD peripheral.
    DEV_MAJOR = dict(zinger=1, minimuffin=2, dingdong=3, hoho=4)

    def _index_firmware_avail(self):
        """Index the various USB-PD firmwares in the rootfs.

        TODO(crosbug.com/434522) This method will need reworked after we've come
        up with a better method for firmware release.

        @returns: dictionary of firmwares (key == name, value == list of
          firmware paths)
        """
        fw_dict = collections.defaultdict(list)
        for fw in glob.glob('%s/*_v[1-9].*.bin' % (self.FW_PATH)):
            mat = re.match(self.FW_NAME_RE, fw)
            if not mat:
                continue

            name = mat.group(1)
            fw_dict[name].append(fw)

        return fw_dict

    def _is_gfu(self, port):
        """Is it in GFU?

        @param port: EC_USBPD object for port.

        @returns: True if GFU enterd, False otherwise.
        """
        return port.is_amode_supported(self.GOOGLE_VID)

    def _is_in_rw(self, port):
        """Is PD device in RW firmware?

        @param port: EC_USBPD object for port.

        @returns: True if in RW, False otherwise.
        """
        flash_info = port.get_flash_info()
        logging.debug('flash_info = %s', flash_info)
        return flash_info['image_status'] == 'RW'

    def _set_kernel_fw_update(self, disable=0):
        """Disable the FW update driver.

        @param disable: 1 for disable, 0 for enable.
        """
        utils.write_one_line(self.FW_UP_DISABLE_PATH, disable)
        if not disable:
            # Allow kernel driver time quiesce
            time.sleep(2)

    def _modify_rw(self, port, rw=None, tries=3):
        """Modify RW of USB-PD device in <port>.

        @param port: EC_USBPD object for port.
        @param rw: Path to RW FW to write using ectool.  If None then uses
          /dev/null to invalidate the RW.
        @param tries: Number of tries to update RW via flashpd

        @returns: True if success, False otherwise.
        """
        timeout = self.MAX_UPDATE_SECS

        if not rw:
            rw = '/dev/null'
            tries = 1

        self._set_kernel_fw_update(disable=1)

        while (tries):
            try:
                # Note in flashpd <dev_major> <port> <file> the dev_major is
                # unnecessary in all cases so its just been set to 0
                port.ec_command('flashpd 0 %d %s' % (port.index, rw),
                                ignore_status=True, timeout=timeout)

            except error.CmdTimeoutError:
                # TODO(tbroch) could remove try/except if ec_command used run
                # instead of system_output + ignore_timeout=True
                tries -= 1
                continue

            if rw != '/dev/null' and not self._is_in_rw(port):
                logging.warn('Port%d: not in RW after flashpd ... retrying',
                             port.index)
                tries -= 1
            else:
                break

        self._set_kernel_fw_update()

        msg = self._reader.get_last_msg([r'%s.*is in RO' % port.index,
                                         self.FW_UP_DNAME],
                                        retries=5, sleep_seconds=2)
        if not msg:
            logging.warn('Port%d: Driver does NOT see dev in not in RO',
                         port.index)
            return False
        logging.info('Port%d: Driver sees device in RO', port.index)
        return True

    def _test_update(self, port, rw=None, tries=3):
        """Test RW update.

        Method tests the kernel's RW update process by first modifying the
        existing RW (either invalidating or rolling it back) via ectool.  It
        then querys the syslog to validate kernel sees the need for update and
        is successful.

        @param port: EC_USBPD object for port.
        @param rw: path to RW firmware to write via ectool to test upgrade.
        @param tries: integer number of attempts to write RW.  Necessary as
          update is not robust (design decision).
        """
        if not tries:
            raise error.TestError('Retries must be > 0')

        if not self._is_in_rw(port):
            raise error.TestError('Port%d: Device is not in RW' % port.index)

        fw_up_re = r'%s.*Port%d FW update completed' % (self.FW_UP_DNAME,
                                                        port.index)

        while tries:
            self._reader.set_start_by_current()
            rsp = self._modify_rw(port, rw)

            if not rsp:
                rsp_str = 'Port%d: RW modified with RW=%s failed' % \
                          (port.index, rw)
                if tries:
                    logging.warn('%s ... retrying.', rsp_str)
                    tries -= 1
                else:
                    raise error.TestError(rsp_str)

            self._reader.set_start_by_current()
            msg = self._reader.get_last_msg([fw_up_re],
                                            retries=(self.MAX_UPDATE_SECS / 2),
                                            sleep_seconds=2)

            if not msg:
                rsp_str = 'Port%d: driver did NOT update FW' % port.index
                if tries:
                    logging.warn('%s ... retrying.', rsp_str)
                    tries -= 1
                    continue
                else:
                    raise error.TestError(rsp_str)

            logging.info('Port%d: Driver completed RW update', port.index)

            # Allow adequate reboot time after RW write completes and device is
            # rebooted.
            time.sleep(3)

            if not self._is_in_rw(port):
                rsp_str = 'Port%d: Device is not in RW' % port.index
                if tries:
                    logging.warn('%s ... retrying.', rsp_str)
                    tries -= 1
                    continue
                else:
                    raise error.TestError(rsp_str)

            break # success #

    def _test_rw_rollback(self, port, fw_dict):
        """Test rolling back RW firmware.

        @param port: EC_USBPD object for port.
        @param fw_dict: dictionary of firmwares.
        """
        self._set_kernel_fw_update()

        # test old RW update
        flash_info = port.get_flash_info()
        for dev_name in fw_dict.keys():
            if flash_info['dev_major'] == self.DEV_MAJOR[dev_name]:
                for old_rw in sorted(fw_dict[dev_name], reverse=True)[1:]:
                    logging.info('Port%d: Rollback test %s to %s',
                                 port.index, dev_name, old_rw)
                    self._test_update(port, rw=old_rw)
                break

    def _test_ro_only(self, port, ro_reps):
        """Test FW update on device with RO only.

        @param port: EC_USBPD object for port.
        @param ro_reps: Number of times to repeat test.
        """
        # test update in RO ro_reps times
        for i in xrange(ro_reps):
            logging.info('RO Loop%d', i)
            self._test_update(port)

    def run_once(self, ro_reps=1):

        fw_dict = self._index_firmware_avail()

        self._usbpd = cros_ec.EC_USBPD()
        self._reader = cros_logging.LogReader()

        for port in self._usbpd.ports:
            if not port.is_dfp():
                continue

            logging.info('Port%d: is a DFP', port.index)

            if not self._is_gfu(port):
                continue

            logging.info('Port%d: supports GFU', port.index)

            self._test_rw_rollback(port, fw_dict)
            self._test_ro_only(port, ro_reps)

    def cleanup(self):
        self._set_kernel_fw_update()