xref: /aosp_15_r20/external/autotest/server/hosts/repair_utils.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7) 
1# Copyright 2016 The Chromium OS Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import logging
6import socket
7
8import common
9from autotest_lib.client.common_lib import hosts
10from autotest_lib.server import utils
11from autotest_lib.server.hosts import servo_constants
12from autotest_lib.server.hosts import cros_constants
13
14from autotest_lib.utils.frozen_chromite.lib import timeout_util
15
16
17def require_servo(host, ignore_state=False):
18    """Require a DUT to have a working servo for a repair action.
19
20    @param host             Host object that require servo.
21    @param ignore_state     Ignore servo state as long as the we still have
22                            servo connection. Some non-critical verifier
23                            failures may not cause servo connection been
24                            disconnected.
25    """
26    servo_initialized = host.servo is not None
27    servo_working = (host.get_servo_state() ==
28                     servo_constants.SERVO_STATE_WORKING or ignore_state)
29
30    if not (servo_initialized and servo_working):
31        raise hosts.AutoservRepairError(
32                '%s has no working servo.' % host.hostname, 'no_working_servo')
33    logging.info('Servo dependence is available for the RepairAction/test.')
34
35
36class SshVerifier(hosts.Verifier):
37    """
38    Verifier to test a host's accessibility via `ssh`.
39
40    This verifier checks whether a given host is reachable over `ssh`.
41    In the event of failure, it distinguishes one of three distinct
42    conditions:
43      * The host can't be found with a DNS lookup.
44      * The host doesn't answer to ping.
45      * The host answers to ping, but not to ssh.
46    """
47
48    @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC)
49    def verify(self, host):
50        if host.is_up():
51            return
52        msg = 'No answer to ssh from %s'
53        try:
54            socket.gethostbyname(host.hostname)
55        except Exception as e:
56            logging.exception('DNS lookup failure')
57            msg = 'Unable to look up %%s in DNS: %s' % e
58        else:
59            if utils.ping(host.hostname, tries=1, deadline=1) != 0:
60                msg = 'No answer to ping from %s'
61        raise hosts.AutoservVerifyError(msg % host.hostname)
62
63    @property
64    def description(self):
65        return 'host is available via ssh'
66
67
68class PingVerifier(hosts.Verifier):
69    """
70    Verifier to test a host's accessibility via `ping`.
71
72    This verifier checks whether a given host is reachable over `ping`.
73    The device is pingable as soon as booted to level when network driver
74    can respond.
75    In the event of failure, it distinguishes one of distinct conditions:
76      * The host can't be found with a DNS lookup.
77      * The host doesn't booted with network drivers.
78    """
79
80    @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC)
81    def verify(self, host):
82        if host.is_up_fast(count=10):
83            return
84        msg = 'No answer to ping to %s'
85        ip_address = None
86        try:
87            ip_address = socket.gethostbyname(host.hostname)
88        except Exception as e:
89            logging.exception('DNS lookup failure')
90            msg = 'Unable to look up %s in DNS: %s' % (host.hostname, str(e))
91            raise hosts.AutoservVerifyError(msg)
92        if not ip_address:
93            msg = 'Hostname: %s not present in DNS' % host.hostname
94        else:
95            msg = 'Hostname: %s not pingable' % host.hostname
96        raise hosts.AutoservVerifyError(msg)
97
98    @property
99    def description(self):
100        return 'host is available via ping'
101
102
103class LegacyHostVerifier(hosts.Verifier):
104    """
105    Ask a Host instance to perform its own verification.
106
107    This class exists as a temporary legacy during refactoring to
108    provide access to code that hasn't yet been rewritten to use the new
109    repair and verify framework.
110    """
111
112    @timeout_util.TimeoutDecorator(cros_constants.VERIFY_TIMEOUT_SEC)
113    def verify(self, host):
114        host.verify_software()
115        host.verify_hardware()
116
117    @property
118    def description(self):
119        return 'Legacy host verification checks'
120
121
122class RebootRepair(hosts.RepairAction):
123    """Repair a target device by rebooting it."""
124
125    # Timeout of this action should defined in child class.
126    def repair(self, host):
127        host.reboot()
128
129    @property
130    def description(self):
131        return 'Reboot the host'
132
133
134class RPMCycleRepair(hosts.RepairAction):
135    """
136    Cycle AC power using the RPM infrastructure.
137
138    This is meant to catch two distinct kinds of failure:
139      * If the target has no battery (that is, a chromebox), power
140        cycling it may force it back on.
141      * If the target has a batter that is discharging or even fully
142        drained, power cycling will leave power on, enabling other
143        repair procedures.
144    """
145
146    @timeout_util.TimeoutDecorator(cros_constants.REPAIR_TIMEOUT_SEC)
147    def repair(self, host):
148        if not host.has_power():
149            raise hosts.AutoservRepairError(
150                    '%s has no RPM connection.' % host.hostname,
151                    'no_working_rpm')
152        host.power_cycle()
153
154
155    @property
156    def description(self):
157        return 'Power cycle the host with RPM'
158