| # Lint as: python2, python3 |
| # Copyright 2016 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. |
| |
| from __future__ import absolute_import |
| from __future__ import division |
| from __future__ import print_function |
| |
| import sys |
| import functools |
| import logging |
| import math |
| import time |
| |
| import common |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.common_lib import hosts |
| from autotest_lib.client.common_lib import utils |
| from autotest_lib.server.cros.servo import servo |
| from autotest_lib.server.hosts import cros_constants |
| from autotest_lib.server.hosts import repair_utils |
| from autotest_lib.server.hosts import servo_constants |
| from autotest_lib.server.cros.servo.topology import servo_topology |
| from autotest_lib.site_utils.admin_audit import servo_updater |
| import six |
| |
| try: |
| from autotest_lib.utils.frozen_chromite.lib import metrics |
| except ImportError: |
| metrics = utils.metrics_mock |
| |
| from autotest_lib.utils.frozen_chromite.lib import timeout_util |
| |
| |
| def ignore_exception_for_non_cros_host(func): |
| """ |
| Decorator to ignore ControlUnavailableError if servo host is not cros host. |
| When using test_that command on a workstation, this enables usage of |
| additional servo devices such as servo micro and Sweetberry. This shall not |
| change any lab behavior. |
| """ |
| @functools.wraps(func) |
| def wrapper(self, host): |
| """ |
| Wrapper around func. |
| """ |
| try: |
| func(self, host) |
| except servo.ControlUnavailableError as e: |
| if host.is_cros_host(): |
| raise |
| logging.warning("Servo host is not cros host, ignore %s: %s", |
| type(e).__name__, e) |
| return wrapper |
| |
| |
| class _UpdateVerifier(hosts.Verifier): |
| """ |
| Verifier to trigger a servo host update, if necessary. |
| |
| The verifier works only for servo_v3. |
| The operation doesn't wait for the update to complete and is |
| considered a success whether or not the servo is currently |
| up-to-date. |
| """ |
| |
| @timeout_util.TimeoutDecorator(cros_constants.LONG_VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| try: |
| if ( |
| not host.get_dut_host_info() |
| or not host.get_dut_host_info().servo_cros_stable_version): |
| logging.info('Servo stable version missed.' |
| ' Skip update check action.') |
| return |
| # We have seen cases that invalid GPT headers/entries block |
| # v3s from been update, so always try to repair here. |
| # See crbug.com/994396, crbug.com/1057302. |
| host.run('cgpt repair /dev/mmcblk0', ignore_status=True) |
| host.update_image() |
| # We don't want failure from update block DUT repair action. |
| # See crbug.com/1029950. |
| except Exception as e: |
| six.reraise(hosts.AutoservNonCriticalVerifyError, str(e), |
| sys.exc_info()[2]) |
| |
| def _is_applicable(self, host): |
| # Run only for servo_v3 host. |
| if host.is_labstation(): |
| return False |
| # Only run if the host is in the physical lab. |
| if not host.is_in_lab() or host.is_localhost(): |
| return False |
| # Skip if the test is being run by test_that, because subnet |
| # restrictions can cause the update to fail. |
| return host.job and host.job.in_lab |
| |
| @property |
| def description(self): |
| return 'Servo_v3 host software is up-to-date' |
| |
| |
| class _StartServodVerifier(hosts.Verifier): |
| """First start of servod on the host. |
| |
| Single running action to start servod in the first time. |
| This verifier created to fit current flow and will be revisited later. |
| Action never fails! |
| """ |
| |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| if not hasattr(self, 'started'): |
| logging.info('Starting servod!') |
| try: |
| host.restart_servod(quick_startup=True) |
| except Exception as e: |
| logging.warning( |
| "Start servod failed due to:\n%s\n" |
| "This error is forgiven here, we will retry" |
| " in further repair actions.", e) |
| # caching the value to prevent restart service when trigger verifier. |
| self.started = True |
| |
| @property |
| def description(self): |
| return 'Initial servod start' |
| |
| |
| class _RootServoPresentVerifier(hosts.Verifier): |
| """Verifier that first servo is present.""" |
| |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| device = None |
| try: |
| topology = host.get_topology() |
| device = topology.get_root_servo() |
| except: |
| host.request_reboot() |
| logging.info('Reboot labstation requested, it will be handled' |
| ' by labstation AdminRepair task.' |
| ' Unable to detect root servo info from topology.') |
| if device and device.is_good(): |
| logging.debug('Root servo is present') |
| else: |
| raise hosts.AutoservVerifyError('Root servo not found!') |
| |
| def _is_applicable(self, host): |
| # Run only for servos under labstations. |
| if not host.is_labstation(): |
| return False |
| # Only run if the host is in the physical lab. |
| if not host.is_in_lab() or host.is_localhost(): |
| return False |
| return True |
| |
| @property |
| def description(self): |
| return 'Root servo is present' |
| |
| |
| class _RootServoV3PresentVerifier(hosts.Verifier): |
| """Verifier that first servo is present.""" |
| |
| RETRY_COUNT = 3 |
| |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| for a in range(self.RETRY_COUNT): |
| logging.debug('Attempt: %s find servo board on servo_v3.', a + 1) |
| present = host.is_servo_board_present_on_servo_v3() |
| if present == False: |
| raise hosts.AutoservVerifyError('Servo board not found!') |
| elif present == True: |
| logging.debug('Servo board is present') |
| return |
| raise hosts.AutoservVerifyError('Fail to find servo board!') |
| |
| def _is_applicable(self, host): |
| # Do not run for servos under labstations. |
| if host.is_labstation(): |
| return False |
| # Only run if the host is in the physical lab. |
| if not host.is_in_lab() or host.is_localhost(): |
| return False |
| return True |
| |
| @property |
| def description(self): |
| return 'Servo board on servo_v3 is present' |
| |
| |
| class _ServoFwVerifier(hosts.Verifier): |
| """Verifier to check is a servo fw is up-to-date.""" |
| |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| try: |
| if servo_updater.any_servo_needs_firmware_update(host): |
| raise hosts.AutoservNonCriticalVerifyError( |
| 'Some servo requires firmware update') |
| except servo_updater.ServoFwVersionMissedError as e: |
| # Do not fail as it will trigger re-flash fw on the servo |
| logging.info( |
| 'Issue with detect new version of firmware for servo.' |
| ' Please file a bug agains Fleet Automation team (go/fleet-bug)' |
| ) |
| |
| def _is_applicable(self, host): |
| # Run only for servos under labstations. |
| if not host.is_labstation(): |
| return False |
| # Only run if the host is in the physical lab. |
| if not host.is_in_lab() or host.is_localhost(): |
| return False |
| return True |
| |
| @property |
| def description(self): |
| return 'Servo fw is up-to-date' |
| |
| |
| class _ConfigVerifier(hosts.Verifier): |
| """ |
| Base verifier for the servo config file verifiers. |
| """ |
| |
| CONFIG_FILE = '/var/lib/servod/config' |
| ATTR = '' |
| |
| @staticmethod |
| def _get_config_val(host, config_file, attr): |
| """ |
| Get the `attr` for `host` from `config_file`. |
| |
| @param host Host to be checked for `config_file`. |
| @param config_file Path to the config file to be tested. |
| @param attr Attribute to get from config file. |
| |
| @return The attr val as set in the config file, or `None` if |
| the file was absent. |
| """ |
| getboard = ('CONFIG=%s ; [ -f $CONFIG ] && ' |
| '. $CONFIG && echo $%s' % (config_file, attr)) |
| attr_val = host.run(getboard, ignore_status=True).stdout |
| return attr_val.strip('\n') if attr_val else None |
| |
| @staticmethod |
| def _validate_attr(host, val, expected_val, attr, config_file): |
| """ |
| Check that the attr setting is valid for the host. |
| |
| This presupposes that a valid config file was found. Raise an |
| execption if: |
| * There was no attr setting from the file (i.e. the setting |
| is an empty string), or |
| * The attr setting is valid, the attr is known, |
| and the setting doesn't match the DUT. |
| |
| @param host Host to be checked for `config_file`. |
| @param val Value to be tested. |
| @param expected_val Expected value. |
| @param attr Attribute we're validating. |
| @param config_file Path to the config file to be tested. |
| """ |
| if not val: |
| raise hosts.AutoservVerifyError( |
| 'config file %s exists, but %s ' |
| 'is not set' % (attr, config_file)) |
| if expected_val is not None and val != expected_val: |
| raise hosts.AutoservVerifyError( |
| '%s is %s; it should be %s' % (attr, val, expected_val)) |
| |
| |
| def _get_config(self, host): |
| """ |
| Return the config file to check. |
| |
| @param host Host object. |
| |
| @return The config file to check. |
| """ |
| return '%s_%d' % (self.CONFIG_FILE, host.servo_port) |
| |
| @property |
| def description(self): |
| return 'servo %s setting is correct' % self.ATTR |
| |
| |
| class _SerialConfigVerifier(_ConfigVerifier): |
| """ |
| Verifier for the servo SERIAL configuration. |
| """ |
| |
| ATTR = 'SERIAL' |
| |
| @timeout_util.TimeoutDecorator(cros_constants.SHORT_VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| """ |
| Test whether the `host` has a `SERIAL` setting configured. |
| |
| This tests the config file names used by the `servod` upstart |
| job for a valid setting of the `SERIAL` variable. The following |
| conditions raise errors: |
| * The SERIAL setting doesn't match the DUT's entry in the AFE |
| database. |
| * There is no config file. |
| """ |
| if not host.is_cros_host(): |
| return |
| # Not all servo hosts will have a servo serial so don't verify if it's |
| # not set. |
| if host.servo_serial is None: |
| return |
| config = self._get_config(host) |
| serialval = self._get_config_val(host, config, self.ATTR) |
| if serialval is None: |
| raise hosts.AutoservVerifyError( |
| 'Servo serial is unconfigured; should be %s' |
| % host.servo_serial |
| ) |
| |
| self._validate_attr(host, serialval, host.servo_serial, self.ATTR, |
| config) |
| |
| |
| |
| class _BoardConfigVerifier(_ConfigVerifier): |
| """ |
| Verifier for the servo BOARD configuration. |
| """ |
| |
| ATTR = 'BOARD' |
| |
| @timeout_util.TimeoutDecorator(cros_constants.SHORT_VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| """ |
| Test whether the `host` has a `BOARD` setting configured. |
| |
| This tests the config file names used by the `servod` upstart |
| job for a valid setting of the `BOARD` variable. The following |
| conditions raise errors: |
| * A config file exists, but the content contains no setting |
| for BOARD. |
| * The BOARD setting doesn't match the DUT's entry in the AFE |
| database. |
| * There is no config file. |
| """ |
| if not host.is_cros_host(): |
| return |
| config = self._get_config(host) |
| boardval = self._get_config_val(host, config, self.ATTR) |
| if boardval is None: |
| msg = 'Servo board is unconfigured' |
| if host.servo_board is not None: |
| msg += '; should be %s' % host.servo_board |
| raise hosts.AutoservVerifyError(msg) |
| |
| self._validate_attr(host, boardval, host.servo_board, self.ATTR, |
| config) |
| |
| |
| class _ServodJobVerifier(hosts.Verifier): |
| """ |
| Verifier to check that the `servod` upstart job is running. |
| """ |
| |
| @timeout_util.TimeoutDecorator(cros_constants.SHORT_VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| if not host.is_cros_host(): |
| return |
| status_cmd = 'status servod PORT=%d' % host.servo_port |
| job_status = host.run(status_cmd, ignore_status=True).stdout |
| if 'start/running' not in job_status: |
| raise hosts.AutoservVerifyError( |
| 'servod not running on %s port %d' % |
| (host.hostname, host.servo_port)) |
| |
| @property |
| def description(self): |
| return 'servod upstart job is running' |
| |
| |
| class _DiskSpaceVerifier(hosts.Verifier): |
| """ |
| Verifier to make sure there is enough disk space left on servohost. |
| """ |
| |
| @timeout_util.TimeoutDecorator(cros_constants.SHORT_VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| # Check available space of stateful is greater than threshold, in Gib. |
| host.check_diskspace('/mnt/stateful_partition', 0.5) |
| |
| @property |
| def description(self): |
| return 'servohost has enough disk space.' |
| |
| |
| class _ServodConnectionVerifier(hosts.Verifier): |
| """ |
| Verifier to check that we can connect to servod server. |
| |
| If this verifier failed, it most likely servod was crashed or in a |
| crashing loop. For servo_v4 it's usually caused by not able to detect |
| CCD or servo_micro. |
| """ |
| |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| host.initilize_servo() |
| |
| @property |
| def description(self): |
| return 'servod service is taking calls' |
| |
| |
| class _ServodControlVerifier(hosts.Verifier): |
| """ |
| Verifier to check basic servo control functionality. |
| |
| This tests the connection to the target servod service with a simple |
| method call. As a side-effect, all servo signals are initialized to |
| default values. |
| |
| N.B. Initializing servo signals is necessary because the power |
| button and lid switch verifiers both test against expected initial |
| values. |
| """ |
| |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| try: |
| host.initialize_dut_for_servo() |
| except Exception as e: |
| six.reraise(hosts.AutoservNonCriticalVerifyError, str(e), |
| sys.exc_info()[2]) |
| |
| @property |
| def description(self): |
| return 'Basic servod control is working' |
| |
| |
| class _Cr50ConsoleVerifier(hosts.Verifier): |
| """Verifier to check if cr50 console is present and working. |
| |
| Validating based by running commands and expect they will not fail. |
| If any command fail then console is not working as expected. |
| """ |
| |
| COMMAND_TO_CHECK_CONSOLE = ( |
| 'cr50_ccd_level', |
| 'cr50_testlab', |
| 'cr50_ccd_state_flags', |
| ) |
| |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| try: |
| for command in self.COMMAND_TO_CHECK_CONSOLE: |
| if host.get_servo().has_control(command): |
| # Response of command is not important. |
| host.get_servo().get(command) |
| except Exception as e: |
| six.reraise(hosts.AutoservNonCriticalVerifyError, str(e), |
| sys.exc_info()[2]) |
| |
| def _is_applicable(self, host): |
| # Only when DUT is running through ccd. |
| # TODO(coconutruben): replace with ccd API when available in servo.py |
| return (host.get_servo() |
| and host.get_servo().get_main_servo_device() == 'ccd_cr50') |
| |
| @property |
| def description(self): |
| return 'CR50 console is working' |
| |
| |
| class _CCDTestlabVerifier(hosts.Verifier): |
| """ |
| Verifier to check that ccd testlab is enabled. |
| |
| All DUT connected by ccd has to supported cr50 with enabled testlab |
| to allow manipulation by servo. The flag testlab is sticky and will |
| stay enabled if was set up. The testlab can be enabled when ccd is |
| open. (go/ccd-setup) |
| """ |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| if not host.get_servo().has_control('cr50_testlab'): |
| raise hosts.AutoservVerifyError( |
| 'cr50 has to be supported when use servo with ' |
| 'ccd_cr50/type-c connection') |
| |
| status = host.get_servo().get('cr50_testlab') |
| # check by 'on' to fail when get unexpected value |
| if status == 'on': |
| # ccd testlab enabled |
| return |
| raise hosts.AutoservNonCriticalVerifyError( |
| 'The ccd testlab is disabled; DUT requires manual work ' |
| 'to enable it (go/ccd-setup).') |
| |
| def _is_applicable(self, host): |
| # Only when DUT is running through ccd. |
| # TODO(coconutruben): replace with ccd API when available in servo.py |
| return (host.get_servo() |
| and host.get_servo().get_main_servo_device() == 'ccd_cr50') |
| |
| @property |
| def description(self): |
| return 'ccd testlab enabled' |
| |
| class _CCDPowerDeliveryVerifier(hosts.Verifier): |
| """Verifier to check and reset servo_v4_role for servos that support |
| power delivery feature(a.k.a power pass through). |
| |
| There are currently two position of servo_v4_role, src and snk: |
| src -- servo in power delivery mode and passes power to the DUT. |
| snk -- servo in normal mode and not passes power to DUT. |
| We want to ensure that servo_v4_role is set to src. |
| """ |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| if host.get_servo().get('servo_pd_role') == 'snk': |
| raise hosts.AutoservNonCriticalVerifyError( |
| 'Power delivery not in src role.') |
| |
| def _is_applicable(self, host): |
| return (host.is_in_lab() and |
| host.get_servo().supports_built_in_pd_control()) |
| |
| @property |
| def description(self): |
| return 'ensure applicable servo is in "src" mode for power delivery' |
| |
| |
| class _BaseDUTConnectionVerifier(hosts.Verifier): |
| """Verifier to check connection between DUT and servo.""" |
| |
| # Bus voltage on ppdut5. Value can be: |
| # - less than 500 - DUT is likely not connected |
| # - between 500 and 4000 - unexpected value |
| # - more than 4000 - DUT is likely connected |
| MAX_PPDUT5_MV_WHEN_NOT_CONNECTED = 500 |
| MIN_PPDUT5_MV_WHEN_CONNECTED = 4000 |
| |
| def _is_usb_hub_connected(self, host): |
| """Checking bus voltage on ppdut5. |
| |
| Supported only on servo_v4 boards. |
| If voltage value is lower than 500 then device is not connected. |
| When value higher 4000 means the device is connected. If value |
| between 500 and 4000 is not expected and will be marked as connected |
| and collected information which DUT has this exception. |
| |
| @returns: bool |
| """ |
| logging.debug('Started check by ppdut5_mv:on') |
| try: |
| val = host.get_servo().get('ppdut5_mv') |
| logging.info('ppdut5_mv=%s', val) |
| if val < self.MAX_PPDUT5_MV_WHEN_NOT_CONNECTED: |
| # servo is not connected to the DUT |
| return False |
| if val < self.MIN_PPDUT5_MV_WHEN_CONNECTED: |
| # is unexpected value. |
| # collecting metrics to look case by case |
| # TODO(otabek) for analysis b:163845694 |
| data = host._get_host_metrics_data() |
| metrics.Counter('chromeos/autotest/repair/ppdut5_mv_case' |
| ).increment(fields=data) |
| # else: |
| # servo is physical connected to the DUT |
| except Exception as e: |
| logging.debug('(Not critical) %s', e) |
| return True |
| |
| def _is_ribbon_cable_connected(self, host): |
| """Check if ribbon cable is connected to the DUT. |
| |
| The servo_micro/flex - can be checked by `cold_reset` signal. |
| When `cold_reset` is `on` it commonly indicates that the DUT |
| is disconnected. To avoid mistake of real signal we try |
| switch it off and if is cannot then servo is not connected. |
| |
| @returns: bool |
| """ |
| logging.debug('Started check by cold_reset:on') |
| try: |
| val = host.get_servo().get('cold_reset') |
| logging.info('cold_reset=%s', val) |
| if val == 'on': |
| # If cold_reset has is on can be right signal |
| # or caused by missing connection between servo_micro and DUT. |
| # if we can switch it to the off then it was signal. |
| host.get_servo().set('cold_reset', 'off') |
| except error.TestFail: |
| logging.debug('Ribbon cable is not connected to the DUT.') |
| return False |
| except Exception as e: |
| logging.debug('(Not critical) %s', e) |
| return True |
| |
| def _is_dut_power_on(self, host): |
| # DUT is running in normal state. |
| # if EC not supported by board then we expect error |
| try: |
| return host.get_servo().get('ec_system_powerstate') == 'S0' |
| except Exception as e: |
| logging.debug('(Not critical) %s', e) |
| return False |
| |
| def _is_servo_v4_type_a(self, host): |
| return host.is_labstation() and host.get_servo().is_servo_v4_type_a() |
| |
| def _is_servo_v4_type_c(self, host): |
| return host.is_labstation() and host.get_servo().is_servo_v4_type_c() |
| |
| def _is_servo_v3(self, host): |
| return not host.is_labstation() |
| |
| |
| class _DUTConnectionVerifier(_BaseDUTConnectionVerifier): |
| """Verifier to check connection Servo to the DUT. |
| |
| Servo_v4 type-a connected to the DUT by: |
| 1) servo_micro - checked by `cold_reset`. |
| Servo_v4 type-c connected to the DUT by: |
| 1) ccd - checked by ppdut5_mv. |
| Servo_v3 connected to the DUT by: |
| 1) legacy servo header - can be checked by `cold_reset`. |
| """ |
| |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| if self._is_servo_v4_type_a(host): |
| if not self._is_ribbon_cable_connected(host): |
| raise hosts.AutoservVerifyError( |
| 'Servo_micro is likely not connected to the DUT.') |
| elif self._is_servo_v4_type_c(host): |
| if (host.get_servo().supports_built_in_pd_control() |
| and not self._is_usb_hub_connected(host)): |
| raise hosts.AutoservVerifyError( |
| 'Servo_v4 is likely not connected to the DUT.') |
| elif self._is_servo_v3(host): |
| if not self._is_ribbon_cable_connected(host): |
| raise hosts.AutoservVerifyError( |
| 'Servo_v3 is likely not connected to the DUT.') |
| |
| @property |
| def description(self): |
| return 'Ensure the Servo connected to the DUT.' |
| |
| |
| class _ServoHubConnectionVerifier(_BaseDUTConnectionVerifier): |
| """Verifier to check connection ServoHub to DUT. |
| |
| Servo_v4 type-a connected to the DUT by: |
| 1) USB hub - checked by ppdut5_mv. |
| """ |
| |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| if self._is_servo_v4_type_a(host): |
| if (self._is_dut_power_on(host) |
| and not self._is_usb_hub_connected(host)): |
| raise hosts.AutoservVerifyError( |
| 'Servo USB hub is likely not connected to the DUT.') |
| |
| def _is_applicable(self, host): |
| if host.is_ec_supported(): |
| return True |
| logging.info('Host does not support EC.') |
| return False |
| |
| @property |
| def description(self): |
| return 'Ensure the Servo HUB connected to the DUT.' |
| |
| |
| class _BaseCr50SBUVerifier(_BaseDUTConnectionVerifier): |
| """Check servod issue related to SBU voltage.""" |
| |
| # Min SBU voltage to detect usb-device |
| SBU_THRESHOLD = 2500.0 |
| # How many times collect SBU voltage to calc AVG value. |
| _TOTAL_CHECK_SBU_VOLTAGE = 10 |
| |
| def _is_applicable(self, host): |
| if host.is_localhost(): |
| logging.info('Target servo is not in a lab,' |
| ' action is not applicable.') |
| return False |
| if not self._is_servo_v4_type_c(host): |
| logging.info('Check support only servo-v4 (type-c),' |
| ' action is not applicable.') |
| return False |
| return True |
| |
| def _is_sbu_voltage_issue(self, host): |
| """Check if servo does not detected by SBU voltage issue.""" |
| command = 'dut_sbu_voltage_float_fault' |
| if host.get_servo().has_control(command): |
| if host.get_servo().get(command) == 'on': |
| return True |
| return False |
| |
| def _get_max_sbu_value(self, host): |
| """Get average voltage on SBU lines.""" |
| servo = host.get_servo() |
| if not servo.has_control('servo_dut_sbu1_mv'): |
| return -1 |
| s1 = 0 |
| s2 = 0 |
| for i in range(self._TOTAL_CHECK_SBU_VOLTAGE): |
| try: |
| sbu1 = int(servo.get('servo_dut_sbu1_mv')) |
| sbu2 = int(servo.get('servo_dut_sbu2_mv')) |
| logging.debug('Attempt:%2d, sbu1 %4d sbu2 %4d', i, sbu1, sbu2) |
| s1 += sbu1 |
| s2 += sbu2 |
| except error.TestFail as e: |
| # This is a nice to have but if reading this fails, it |
| # shouldn't interfere with the test. |
| logging.exception(e) |
| logging.debug('Total: sbu1 %4d sbu2 %4d', s1, s2) |
| # Use float to get values with changes |
| s1 = s1 / float(self._TOTAL_CHECK_SBU_VOLTAGE) |
| s2 = s2 / float(self._TOTAL_CHECK_SBU_VOLTAGE) |
| logging.debug('Avg: sbu1 %7.2f sbu2 %7.2f', s1, s2) |
| max_sbu = max(s1, s2) |
| logging.info('Max sbu: %7.2f', max_sbu) |
| return max_sbu |
| |
| |
| class _Cr50OffVerifier(_BaseCr50SBUVerifier): |
| """Check if CR50 is in deep sleep and fail to detected. |
| |
| If SBU voltage is higher threshold but still cannot be detected |
| as usb device then probably CR50 is in deep sleep. |
| Threshold is 2500 mV on any SBU lines. |
| """ |
| |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| if self._is_sbu_voltage_issue(host): |
| if self._get_max_sbu_value(host) > self.SBU_THRESHOLD: |
| raise hosts.AutoservVerifyError( |
| 'CR50 voltage detected but usb device not enumerated') |
| |
| @property |
| def description(self): |
| return 'CR50 voltage detected but not enumerated.' |
| |
| |
| class _Cr50LowSBUVerifier(_BaseCr50SBUVerifier): |
| """Check if servod fail to detect CR50 due low voltage. |
| |
| CR50 cannot be enumerated as SBU voltage line lower then |
| threshold. |
| Threshold is 2500 mV on any SBU lines. |
| """ |
| |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| if self._is_sbu_voltage_issue(host): |
| v = self._get_max_sbu_value(host) |
| if v > 1 and v <= self.SBU_THRESHOLD: |
| raise hosts.AutoservVerifyError( |
| 'Cr50 is not detected due to SBU voltages' |
| ' being below %dmV' % self.SBU_THRESHOLD) |
| |
| @property |
| def description(self): |
| return 'Cr50 not detected as both SBU voltages are below threshold.' |
| |
| |
| class _TopologyVerifier(hosts.Verifier): |
| """Verifier that all servo component is presented.""" |
| |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| topology = host.get_topology() |
| topology.read(host.get_dut_host_info()) |
| try: |
| # Linux takes 1 second to detect and enumerate USB device since |
| # 2010 year. We take 10 seconds to be sure as old standard was |
| # 5 seconds. |
| time.sleep(10) |
| topology.validate(raise_error=True, |
| dual_set=host.is_dual_setup(), |
| compare=True) |
| except servo_topology.ServoTopologyError as e: |
| six.reraise(hosts.AutoservVerifyError, str(e), sys.exc_info()[2]) |
| |
| def _is_applicable(self, host): |
| if host.is_localhost(): |
| logging.info('Target servo is not in a lab,' |
| ' action is not applicable.') |
| return False |
| if not host.is_servo_topology_supported(): |
| logging.info('Target servo-topology is not supported,' |
| ' action is not applicable.') |
| return False |
| return True |
| |
| @property |
| def description(self): |
| return 'Ensure all Servo component present.' |
| |
| |
| class _PowerButtonVerifier(hosts.Verifier): |
| """ |
| Verifier to check the `pwr_button` signal. |
| |
| Tests that the `pwr_button` signal shows the power button has been |
| released. When `pwr_button` is stuck at `press`, it commonly |
| indicates that the ribbon cable is disconnected. |
| """ |
| # TODO (crbug.com/646593) - Remove list below once servo has been updated |
| # with a fake pwr_button signal. |
| _BOARDS_WO_PWR_BUTTON = ['arkham', 'gale', 'mistral', 'storm', 'whirlwind'] |
| |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.SHORT_VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| if host.servo_board in self._BOARDS_WO_PWR_BUTTON: |
| return |
| try: |
| button = host.get_servo().get('pwr_button') |
| except Exception as e: |
| six.reraise(hosts.AutoservNonCriticalVerifyError, str(e), |
| sys.exc_info()[2]) |
| |
| if button != 'release': |
| raise hosts.AutoservNonCriticalVerifyError( |
| 'Check ribbon cable: \'pwr_button\' is stuck') |
| |
| def _is_applicable(self, host): |
| return (host.get_servo() and host.get_servo().main_device_is_flex()) |
| |
| @property |
| def description(self): |
| return 'pwr_button control is normal' |
| |
| |
| class _BatteryVerifier(hosts.Verifier): |
| """Collect battery info for analysis.""" |
| |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| try: |
| servo = host.get_servo() |
| charging = False |
| if servo.has_control('battery_is_charging'): |
| charging = servo.get('battery_is_charging') |
| level = -1 |
| if servo.has_control('battery_charge_percent'): |
| level = servo.get('battery_charge_percent') |
| design_mah = servo.get('battery_full_design_mah') |
| charge_mah = servo.get('battery_full_charge_mah') |
| logging.info('Charging: %s', charging) |
| logging.info('Percentage: %s', level) |
| logging.info('Full charge max: %s', charge_mah) |
| logging.info('Full design max: %s', design_mah) |
| # based on analysis of ratio we can find out what is |
| # the level when we can say that battery is dead |
| ratio = int(math.floor(charge_mah / design_mah * 100.0)) |
| logging.info('Ratio: %s', ratio) |
| data = { |
| 'board': host.servo_board or 'unknown', |
| 'model': host.servo_model or 'unknown', |
| 'ratio': ratio |
| } |
| metrics.Counter('chromeos/autotest/battery/ratio').increment( |
| fields=data) |
| except Exception as e: |
| # Keeping it with info level because we do not expect it. |
| logging.info('(Not critical) %s', e) |
| |
| def _is_applicable(self, host): |
| if not host.is_ec_supported(): |
| logging.info('The board not support EC') |
| return False |
| dut_info = host.get_dut_host_info() |
| if dut_info: |
| host_info = host.get_dut_host_info() |
| if host_info.get_label_value('power') != 'battery': |
| logging.info('The board does not have battery') |
| return False |
| servo = host.get_servo() |
| if (not servo.has_control('battery_full_design_mah') |
| or not servo.has_control('battery_full_charge_mah')): |
| logging.info('The board is not supported battery controls...') |
| return False |
| return True |
| |
| @property |
| def description(self): |
| return 'Logs battery levels' |
| |
| |
| class _LidVerifier(hosts.Verifier): |
| """ |
| Verifier to check the `lid_open` signal. |
| """ |
| |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.SHORT_VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| try: |
| lid_open = host.get_servo().get('lid_open') |
| except Exception as e: |
| six.reraise(hosts.AutoservNonCriticalVerifyError, str(e), |
| sys.exc_info()[2]) |
| |
| if lid_open != 'yes' and lid_open != 'not_applicable': |
| raise hosts.AutoservNonCriticalVerifyError( |
| 'Check lid switch: lid_open is %s' % lid_open) |
| |
| @property |
| def description(self): |
| return 'lid_open control is normal' |
| |
| |
| class _EcBoardVerifier(hosts.Verifier): |
| """ |
| Verifier response from the 'ec_board' control. |
| """ |
| |
| @ignore_exception_for_non_cros_host |
| @timeout_util.TimeoutDecorator(cros_constants.SHORT_VERIFY_TIMEOUT_SEC) |
| def verify(self, host): |
| if host.is_ec_supported(): |
| ec_board_name = '' |
| try: |
| ec_board_name = host.get_servo().get_ec_board() |
| logging.debug('EC board: %s', ec_board_name) |
| except Exception as e: |
| raise hosts.AutoservNonCriticalVerifyError( |
| '`ec_board` control is not responding; ' |
| 'may be caused of broken EC firmware') |
| else: |
| logging.info('The board not support EC') |
| |
| @property |
| def description(self): |
| return 'Check EC by get `ec_board` control' |
| |
| |
| class _RestartServod(hosts.RepairAction): |
| """Restart `servod` with the proper BOARD setting.""" |
| |
| @timeout_util.TimeoutDecorator(cros_constants.REPAIR_TIMEOUT_SEC) |
| def repair(self, host): |
| if not host.is_cros_host(): |
| raise hosts.AutoservRepairError( |
| 'Can\'t restart servod: not running ' |
| 'embedded Chrome OS.', |
| 'servo_not_applicable_to_non_cros_host') |
| host.restart_servod() |
| |
| @property |
| def description(self): |
| return 'Start servod with the proper config settings.' |
| |
| |
| class _ServoRebootRepair(repair_utils.RebootRepair): |
| """Try repair servo by reboot servohost. |
| |
| This is the same as the standard `RebootRepair`, for servo_v3 it will |
| reboot the beaglebone board immidiately while for labstation it will |
| request a reboot by touch a flag file on its labstation, then |
| labstation reboot will be handled by labstation AdminRepair task as |
| labstation host multiple servos and need do an synchronized reboot. |
| """ |
| |
| @timeout_util.TimeoutDecorator(cros_constants.REPAIR_TIMEOUT_SEC) |
| def repair(self, host): |
| super(_ServoRebootRepair, self).repair(host) |
| # restart servod for v3 after reboot. |
| host.restart_servod() |
| |
| def _is_applicable(self, host): |
| if host.is_localhost() or not host.is_cros_host(): |
| logging.info('Target servo is not in a lab, the reboot repair' |
| ' action is not applicable.') |
| return False |
| |
| if host.is_labstation(): |
| host.request_reboot() |
| logging.info('Reboot labstation requested, it will be handled' |
| ' by labstation AdminRepair task.') |
| return False |
| return True |
| |
| @property |
| def description(self): |
| return 'Reboot the servo host.' |
| |
| |
| class _ToggleCCLineRepair(hosts.RepairAction): |
| """Try repair servod by toggle cc. |
| |
| When cr50 is not enumerated we can try to recover it by toggle cc line. |
| """ |
| # Timeout for shut down configuration channel. |
| CC_OFF_TIMEOUT = 10 |
| # Timeout for initialize configuration channel. |
| CC_ON_TIMEOUT = 30 |
| |
| @timeout_util.TimeoutDecorator(cros_constants.REPAIR_TIMEOUT_SEC) |
| def repair(self, host): |
| logging.info('Turn off configuration channel and wait 10 seconds.') |
| host.get_servo().set_nocheck('servo_v4_uart_cmd', 'cc off') |
| # wait till command will be effected |
| time.sleep(self.CC_OFF_TIMEOUT) |
| |
| logging.info('Turn on configuration channel and wait 30 seconds.') |
| # alternative option to turn line on is by `cc srcdts` |
| host.get_servo().set_nocheck('servo_pd_role', 'src') |
| host.get_servo().set_nocheck('servo_dts_mode', 'on') |
| # wait till command will be effected |
| time.sleep(self.CC_ON_TIMEOUT) |
| host.restart_servod() |
| |
| def _is_applicable(self, host): |
| if host.is_localhost() or not host.is_labstation(): |
| return False |
| if not host.servo_serial: |
| return False |
| if not host.servo_recovery: |
| logging.debug('Servod is not running in recovery mode.') |
| return False |
| if not host.get_servo(): |
| logging.debug('Servo is not initialized.') |
| return False |
| return self._is_type_c(host) |
| |
| def _is_type_c(self, host): |
| if host.get_dut_host_info(): |
| servo_type = host.get_dut_host_info().get_label_value( |
| servo_constants.SERVO_TYPE_LABEL_PREFIX) |
| return 'ccd_cr50' in servo_type |
| return False |
| |
| @property |
| def description(self): |
| return 'Toggle cc lines' |
| |
| |
| class _FakedisconnectRepair(hosts.RepairAction): |
| """Try repair servod by mimic reconnection of servo. |
| |
| When cr50 is not enumerated as we can try to recover it by reconnect to DUT. |
| """ |
| # Delay to disconnect. |
| DISC_DELAY_MS = 100 |
| # Timeout to wait to restore the connection. |
| DISC_TIMEOUT_MS = 2000 |
| # Timeout to wait to execute the command and apply effect. |
| EXEC_TIMEOUT = (DISC_DELAY_MS + DISC_TIMEOUT_MS) / 1000 + 2 |
| |
| @timeout_util.TimeoutDecorator(cros_constants.REPAIR_TIMEOUT_SEC) |
| def repair(self, host): |
| disc_cmd = ('fakedisconnect %d %d' % |
| (self.DISC_DELAY_MS, self.DISC_TIMEOUT_MS)) |
| # cannot use 'set' as control is not returned executed commands |
| host.get_servo().set_nocheck('servo_v4_uart_cmd', disc_cmd) |
| logging.debug('Waiting %ss for affect of action', self.EXEC_TIMEOUT) |
| time.sleep(self.EXEC_TIMEOUT) |
| host.restart_servod() |
| |
| def _is_applicable(self, host): |
| if host.is_localhost() or not host.is_labstation(): |
| return False |
| if not host.servo_serial: |
| return False |
| if not host.servo_recovery: |
| logging.debug('Servod is not running in recovery mode.') |
| return False |
| if not host.get_servo(): |
| logging.debug('Servo is not initialized.') |
| return False |
| return self._is_type_c(host) |
| |
| def _is_type_c(self, host): |
| if host.get_dut_host_info(): |
| servo_type = host.get_dut_host_info().get_label_value( |
| servo_constants.SERVO_TYPE_LABEL_PREFIX) |
| return 'ccd_cr50' in servo_type |
| return False |
| |
| @property |
| def description(self): |
| return 'Fake reconnect to DUT' |
| |
| |
| class _PowerDeliveryRepair(hosts.RepairAction): |
| """Repair to check servo_v4_role for servos that support |
| power delivery feature(a.k.a power pass through). |
| |
| There are currently two position of servo_v4_role, src and snk: |
| src -- servo in power delivery mode and passes power to the DUT. |
| snk -- servo in normal mode and not passes power to DUT. |
| """ |
| # How many time retry to set PD in correct mode and verify that is stay. |
| # Set 5 as each attempt has 10 attempts inside 'set' method. |
| _SET_ATTEMPT_COUNT = 5 |
| |
| @timeout_util.TimeoutDecorator(cros_constants.REPAIR_TIMEOUT_SEC) |
| def repair(self, host): |
| host.get_servo().set_nocheck('servo_pd_role', 'snk') |
| time.sleep(1) |
| for x in range(self._SET_ATTEMPT_COUNT): |
| logging.debug('Try set servo_v4_role to src.' |
| ' Attempt: %s', x + 1) |
| try: |
| host.get_servo().set('servo_pd_role', 'src') |
| # Waiting a few seconds as it can be change to snk if PD |
| # on servo has issue. |
| time.sleep(5) |
| except BaseException as e: |
| logging.debug('Setting PD with retries failed %s', e) |
| if host.get_servo().get('servo_pd_role') == 'src': |
| break |
| if host.get_servo().get('servo_pd_role') == 'snk': |
| raise hosts.AutoservNonCriticalVerifyError( |
| 'Cannot switch power delivery to the src role') |
| # Restart servod to re-initialize servos. |
| # In some cases if device did not receive power can block detection |
| # of servo components. |
| host.restart_servod() |
| |
| def _is_type_c(self, host): |
| return (host.is_in_lab() and host.get_servo() |
| and host.get_servo().supports_built_in_pd_control()) |
| |
| @property |
| def description(self): |
| return 'Recover power delivery on servo' |
| |
| |
| class _ECRebootRepair(hosts.RepairAction): |
| """ |
| Reboot EC on DUT from servo. |
| """ |
| |
| def _is_applicable(self, host): |
| return (not host.is_localhost()) and host.is_ec_supported() |
| |
| @timeout_util.TimeoutDecorator(cros_constants.REPAIR_TIMEOUT_SEC) |
| def repair(self, host): |
| host.get_servo().ec_reboot() |
| |
| @property |
| def description(self): |
| return 'Reboot EC' |
| |
| |
| class _DutRebootRepair(hosts.RepairAction): |
| """ |
| Reboot DUT to recover some servo controls depending on EC console. |
| |
| Some servo controls, like lid_open, requires communicating with DUT through |
| EC UART console. Failure of this kinds of controls can be recovered by |
| rebooting the DUT. |
| """ |
| |
| @timeout_util.TimeoutDecorator(cros_constants.REPAIR_TIMEOUT_SEC) |
| def repair(self, host): |
| host.get_servo().get_power_state_controller().reset() |
| # Get the lid_open value which requires EC console. |
| lid_open = host.get_servo().get('lid_open') |
| if lid_open != 'yes' and lid_open != 'not_applicable': |
| raise hosts.AutoservVerifyError( |
| 'Still fail to contact EC console after rebooting DUT') |
| |
| @property |
| def description(self): |
| return 'Reset the DUT via servo' |
| |
| |
| class _DiskCleanupRepair(hosts.RepairAction): |
| """ |
| Remove old logs/metrics/crash_dumps on servohost to free up disk space. |
| """ |
| KEEP_LOGS_MAX_DAYS = 5 |
| |
| FILE_TO_REMOVE = [ |
| '/var/lib/metrics/uma-events', '/var/spool/crash/*', |
| '/var/log/chrome/*', '/var/log/ui/*', |
| '/home/chronos/BrowserMetrics/*' |
| ] |
| |
| @timeout_util.TimeoutDecorator(cros_constants.SHORT_REPAIR_TIMEOUT_SEC) |
| def repair(self, host): |
| if host.is_localhost(): |
| # we don't want to remove anything from local testing. |
| return |
| |
| # Remove old servod logs. |
| host.run('/usr/bin/find /var/log/servod_* -mtime +%d -print -delete' |
| % self.KEEP_LOGS_MAX_DAYS, ignore_status=True) |
| |
| # Remove pre-defined metrics and crash dumps. |
| for path in self.FILE_TO_REMOVE: |
| host.run('rm %s' % path, ignore_status=True) |
| |
| @property |
| def description(self): |
| return 'Clean up old logs/metrics on servohost to free up disk space.' |
| |
| |
| class _ServoFwUpdateRepair(hosts.RepairAction): |
| """Update firmware for servos. |
| |
| We try to update servo 3 times and then try to force update it. |
| """ |
| |
| @timeout_util.TimeoutDecorator(cros_constants.REPAIR_TIMEOUT_SEC) |
| def repair(self, host): |
| servo_updater.update_servo_firmware(host, |
| try_attempt_count=3, |
| force_update=False, |
| try_force_update=True) |
| |
| def _is_applicable(self, host): |
| # Run only for servo_v4 and servo_v4p1. |
| return host.is_labstation() |
| |
| @property |
| def description(self): |
| return 'Update servo-fw if required.' |
| |
| |
| class _ServoMicroFlashRepair(hosts.RepairAction): |
| """ |
| Remove old logs/metrics/crash_dumps on servohost to free up disk space. |
| """ |
| _TARGET_SERVO = 'servo_micro' |
| |
| @timeout_util.TimeoutDecorator(cros_constants.REPAIR_TIMEOUT_SEC) |
| def repair(self, host): |
| if not host.is_cros_host(): |
| raise hosts.AutoservRepairError( |
| 'Can\'t restart servod: not running ' |
| 'embedded Chrome OS.', |
| 'servo_not_applicable_to_non_cros_host') |
| servo = host.get_servo() |
| if not servo or self._TARGET_SERVO not in servo.get_servo_type(): |
| logging.info("Servo-micro is not present on set-up") |
| return |
| |
| try: |
| servo_updater.update_servo_firmware(host, |
| boards=(self._TARGET_SERVO, ), |
| force_update=True, |
| ignore_version=True) |
| except Exception as e: |
| logging.debug("(Not critical) Servo device update error: %s", e) |
| raise hosts.AutoservVerifyError( |
| 'Still fail to contact EC console after rebooting DUT') |
| # Update time when we reflashed the fw on the device |
| dhp = host.get_dut_health_profile() |
| dhp.refresh_servo_miro_fw_update_run_time() |
| host.restart_servod() |
| |
| def is_time_to_try(self, dhp): |
| """Verify that it is time when we can try to re-flash fw on servo_micro. |
| |
| Re-flashing limited to once per 2 weeks to avoid over-flashing |
| the servo device. |
| """ |
| today_time = int(time.time()) |
| last_check = dhp.get_servo_micro_fw_update_time_epoch() |
| can_run = today_time > (last_check + (14 * 24 * 60 * 60)) |
| if not can_run: |
| logging.info("The servo_micro fw updated in las 2 weeks ago.") |
| return can_run |
| |
| def _is_applicable(self, host): |
| return (not host.is_localhost() and host.get_dut_health_profile() |
| and self.is_time_to_try(host.get_dut_health_profile())) |
| |
| @property |
| def description(self): |
| return 'Re-flash servo_micro firmware.' |
| |
| |
| def _servo_verifier_actions(): |
| """ |
| Return a verifiers for a `ServoHost`. |
| """ |
| return ( |
| (repair_utils.SshVerifier, 'servo_ssh', []), |
| (_RootServoPresentVerifier, 'servo_root_present', ['servo_ssh']), |
| (_RootServoV3PresentVerifier, 'servo_v3_root_present', |
| ['servo_ssh']), |
| (_ServoFwVerifier, 'servo_fw', ['servo_root_present']), |
| (_StartServodVerifier, 'start_servod', |
| ['servo_fw', 'servo_v3_root_present']), |
| (_DiskSpaceVerifier, 'servo_disk_space', ['servo_ssh']), |
| (_UpdateVerifier, 'servo_update', ['servo_v3_root_present']), |
| (_BoardConfigVerifier, 'servo_config_board', ['servo_ssh']), |
| (_SerialConfigVerifier, 'servo_config_serial', ['servo_ssh']), |
| (_ServodJobVerifier, 'servod_started', [ |
| 'start_servod', 'servo_v3_root_present', |
| 'servo_config_board', 'servo_config_serial', |
| 'servo_disk_space' |
| ]), |
| (_TopologyVerifier, 'servo_topology', ['servod_started']), |
| (_ServodConnectionVerifier, 'servod_connection', |
| ['servod_started']), |
| (_Cr50LowSBUVerifier, 'servo_cr50_low_sbu', ['servod_connection']), |
| (_Cr50OffVerifier, 'servo_cr50_off', ['servod_connection']), |
| (_ServodControlVerifier, 'servod_control', ['servod_connection']), |
| (_DUTConnectionVerifier, 'servo_dut_connected', |
| ['servod_connection']), |
| (_ServoHubConnectionVerifier, 'servo_hub_connected', |
| ['servo_dut_connected']), |
| (_PowerButtonVerifier, 'servo_pwr_button', ['servo_hub_connected' |
| ]), |
| (_BatteryVerifier, 'servo_battery', ['servo_hub_connected']), |
| (_LidVerifier, 'servo_lid_open', ['servo_hub_connected']), |
| (_EcBoardVerifier, 'servo_ec_board', ['servo_dut_connected']), |
| (_Cr50ConsoleVerifier, 'servo_cr50_console', |
| ['servo_dut_connected']), |
| (_CCDTestlabVerifier, 'servo_ccd_testlab', ['servo_cr50_console']), |
| (_CCDPowerDeliveryVerifier, 'servo_power_delivery', |
| ['servod_connection']), |
| ) |
| |
| |
| def _servo_repair_actions(): |
| """ |
| Return a `RepairStrategy` for a `ServoHost`. |
| """ |
| config = ['servo_config_board', 'servo_config_serial'] |
| base_triggers = [ |
| 'servod_started', 'servo_topology', 'servod_connection', |
| 'servod_control', 'servo_dut_connected', 'servo_hub_connected', |
| 'servo_pwr_button', 'servo_cr50_console', 'servo_cr50_low_sbu', |
| 'servo_cr50_off', 'servo_power_delivery' |
| ] |
| dut_triggers = [ |
| 'servod_control', 'servo_lid_open', 'servo_ec_board', |
| 'servo_topology', 'servo_dut_connected', 'servo_hub_connected', |
| 'servo_cr50_low_sbu', 'servo_cr50_off', 'servo_cr50_console', |
| 'servo_power_delivery' |
| ] |
| return ( |
| (_ServoFwUpdateRepair, 'servo_fw_update', ['servo_ssh'], |
| ['servo_fw']), |
| (_DiskCleanupRepair, 'servo_disk_cleanup', ['servo_ssh'], |
| ['servo_disk_space']), |
| (_ServoMicroFlashRepair, 'servo_micro_flash', |
| ['servo_ssh', 'servo_topology'], ['servo_dut_connected']), |
| (_RestartServod, 'servod_restart', ['servo_ssh'], |
| config + base_triggers), |
| (_ServoRebootRepair, 'servo_reboot', ['servo_ssh'], base_triggers), |
| (_PowerDeliveryRepair, 'servo_pd_recover', ['servod_connection'], |
| base_triggers), |
| (_FakedisconnectRepair, 'servo_fakedisconnect', |
| ['servod_connection'], base_triggers), |
| (_ToggleCCLineRepair, 'servo_cc', ['servod_connection'], |
| base_triggers), |
| (_DutRebootRepair, 'servo_dut_reboot', ['servod_connection'], |
| dut_triggers), |
| (_ECRebootRepair, 'servo_ec_reboot', ['servod_connection'], |
| dut_triggers), |
| ) |
| |
| |
| def create_servo_repair_strategy(): |
| """ |
| Return a `RepairStrategy` for a `ServoHost`. |
| """ |
| return hosts.RepairStrategy(_servo_verifier_actions(), |
| _servo_repair_actions(), 'servo') |
