blob: 89539f68cdf62cff50d891d4757898c8704ba849 [file] [log] [blame]
# 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.
import logging
import socket
import common
from autotest_lib.client.common_lib import hosts
from autotest_lib.server import utils
from autotest_lib.server.hosts import servo_constants
from autotest_lib.server.hosts import cros_constants
from chromite.lib import timeout_util
def require_servo(host, ignore_state=False):
"""Require a DUT to have a working servo for a repair action.
@param host Host object that require servo.
@param ignore_state Ignore servo state as long as the we still have
servo connection. Some non-critical verifier
failures may not cause servo connection been
servo_initialized = host.servo is not None
servo_working = (host.get_servo_state() ==
servo_constants.SERVO_STATE_WORKING or ignore_state)
if not (servo_initialized and servo_working):
raise hosts.AutoservRepairError(
'%s has no working servo.' % host.hostname, 'no_working_servo')'Servo dependence is available for the RepairAction/test.')
class SshVerifier(hosts.Verifier):
Verifier to test a host's accessibility via `ssh`.
This verifier checks whether a given host is reachable over `ssh`.
In the event of failure, it distinguishes one of three distinct
* The host can't be found with a DNS lookup.
* The host doesn't answer to ping.
* The host answers to ping, but not to ssh.
def verify(self, host):
if host.is_up():
msg = 'No answer to ssh from %s'
except Exception as e:
logging.exception('DNS lookup failure')
msg = 'Unable to look up %%s in DNS: %s' % e
if, tries=1, deadline=1) != 0:
msg = 'No answer to ping from %s'
raise hosts.AutoservVerifyError(msg % host.hostname)
def description(self):
return 'host is available via ssh'
class LegacyHostVerifier(hosts.Verifier):
Ask a Host instance to perform its own verification.
This class exists as a temporary legacy during refactoring to
provide access to code that hasn't yet been rewritten to use the new
repair and verify framework.
def verify(self, host):
def description(self):
return 'Legacy host verification checks'
class RebootRepair(hosts.RepairAction):
"""Repair a target device by rebooting it."""
# Timeout of this action should defined in child class.
def repair(self, host):
def description(self):
return 'Reboot the host'
class RPMCycleRepair(hosts.RepairAction):
Cycle AC power using the RPM infrastructure.
This is meant to catch two distinct kinds of failure:
* If the target has no battery (that is, a chromebox), power
cycling it may force it back on.
* If the target has a batter that is discharging or even fully
drained, power cycling will leave power on, enabling other
repair procedures.
def repair(self, host):
if not host.has_power():
raise hosts.AutoservRepairError(
'%s has no RPM connection.' % host.hostname,
def description(self):
return 'Power cycle the host with RPM'