Home | History | Annotate | Download | only in hosts
      1 #!/usr/bin/python
      2 # Copyright 2017 The Chromium OS Authors. All rights reserved.
      3 # Use of this source code is governed by a BSD-style license that can be
      4 # found in the LICENSE file.
      5 
      6 import unittest
      7 
      8 import common
      9 from autotest_lib.server.hosts import cros_firmware
     10 from autotest_lib.server.hosts import cros_repair
     11 from autotest_lib.server.hosts import repair
     12 
     13 
     14 CROS_VERIFY_DAG = (
     15     (repair.SshVerifier, 'ssh', ()),
     16     (cros_repair.DevModeVerifier, 'devmode', ('ssh',)),
     17     (cros_repair.HWIDVerifier,    'hwid',    ('ssh',)),
     18     (cros_repair.ACPowerVerifier, 'power', ('ssh',)),
     19     (cros_repair.EXT4fsErrorVerifier, 'ext4', ('ssh',)),
     20     (cros_repair.WritableVerifier, 'writable', ('ssh',)),
     21     (cros_repair.TPMStatusVerifier, 'tpm', ('ssh',)),
     22     (cros_repair.UpdateSuccessVerifier, 'good_au', ('ssh',)),
     23     (cros_firmware.FirmwareStatusVerifier, 'fwstatus', ('ssh',)),
     24     (cros_firmware.FirmwareVersionVerifier, 'rwfw', ('ssh',)),
     25     (cros_repair.PythonVerifier, 'python', ('ssh',)),
     26     (repair.LegacyHostVerifier, 'cros', ('ssh',)),
     27 )
     28 
     29 CROS_REPAIR_ACTIONS = (
     30     (repair.RPMCycleRepair, 'rpm', (), ('ssh', 'power',)),
     31     (cros_repair.ServoSysRqRepair, 'sysrq', (), ('ssh',)),
     32     (cros_repair.ServoResetRepair, 'servoreset', (), ('ssh',)),
     33     (cros_firmware.FirmwareRepair,
     34      'firmware', (), ('ssh', 'fwstatus', 'good_au')),
     35     (cros_repair.CrosRebootRepair,
     36      'reboot', ('ssh',), ('devmode', 'writable',)),
     37     (cros_repair.AutoUpdateRepair,
     38      'au',
     39      ('ssh', 'writable', 'tpm', 'good_au', 'ext4'),
     40      ('power', 'rwfw', 'python', 'cros')),
     41     (cros_repair.PowerWashRepair,
     42      'powerwash',
     43      ('ssh', 'writable'),
     44      ('tpm', 'good_au', 'ext4', 'power', 'rwfw', 'python', 'cros')),
     45     (cros_repair.ServoInstallRepair,
     46      'usb',
     47      (),
     48      ('ssh', 'writable', 'tpm', 'good_au', 'ext4', 'power', 'rwfw',
     49       'python', 'cros')),
     50 )
     51 
     52 MOBLAB_VERIFY_DAG = (
     53     (repair.SshVerifier, 'ssh', ()),
     54     (cros_repair.ACPowerVerifier, 'power', ('ssh',)),
     55     (cros_firmware.FirmwareVersionVerifier, 'rwfw', ('ssh',)),
     56     (cros_repair.PythonVerifier, 'python', ('ssh',)),
     57     (repair.LegacyHostVerifier, 'cros', ('ssh',)),
     58 )
     59 
     60 MOBLAB_REPAIR_ACTIONS = (
     61     (repair.RPMCycleRepair, 'rpm', (), ('ssh', 'power',)),
     62     (cros_repair.AutoUpdateRepair,
     63      'au', ('ssh',), ('power', 'rwfw', 'python', 'cros',)),
     64 )
     65 
     66 JETSTREAM_VERIFY_DAG = (
     67     (repair.SshVerifier, 'ssh', ()),
     68     (cros_repair.DevModeVerifier, 'devmode', ('ssh',)),
     69     (cros_repair.HWIDVerifier,    'hwid',    ('ssh',)),
     70     (cros_repair.ACPowerVerifier, 'power', ('ssh',)),
     71     (cros_repair.EXT4fsErrorVerifier, 'ext4', ('ssh',)),
     72     (cros_repair.WritableVerifier, 'writable', ('ssh',)),
     73     (cros_repair.TPMStatusVerifier, 'tpm', ('ssh',)),
     74     (cros_repair.UpdateSuccessVerifier, 'good_au', ('ssh',)),
     75     (cros_firmware.FirmwareStatusVerifier, 'fwstatus', ('ssh',)),
     76     (cros_firmware.FirmwareVersionVerifier, 'rwfw', ('ssh',)),
     77     (cros_repair.PythonVerifier, 'python', ('ssh',)),
     78     (repair.LegacyHostVerifier, 'cros', ('ssh',)),
     79     (cros_repair.JetstreamServicesVerifier, 'jetstream_services', ('ssh',)),
     80 )
     81 
     82 JETSTREAM_REPAIR_ACTIONS = (
     83     (repair.RPMCycleRepair, 'rpm', (), ('ssh', 'power',)),
     84     (cros_repair.ServoSysRqRepair, 'sysrq', (), ('ssh',)),
     85     (cros_repair.ServoResetRepair, 'servoreset', (), ('ssh',)),
     86     (cros_firmware.FirmwareRepair,
     87      'firmware', (), ('ssh', 'fwstatus', 'good_au')),
     88     (cros_repair.CrosRebootRepair,
     89      'reboot', ('ssh',), ('devmode', 'writable',)),
     90     (cros_repair.JetstreamRepair,
     91      'jetstream_repair',
     92      ('ssh', 'writable', 'tpm', 'good_au', 'ext4'),
     93      ('power', 'rwfw', 'python', 'cros', 'jetstream_services')),
     94     (cros_repair.AutoUpdateRepair,
     95      'au',
     96      ('ssh', 'writable', 'tpm', 'good_au', 'ext4'),
     97      ('power', 'rwfw', 'python', 'cros', 'jetstream_services')),
     98     (cros_repair.PowerWashRepair,
     99      'powerwash',
    100      ('ssh', 'writable'),
    101      ('tpm', 'good_au', 'ext4', 'power', 'rwfw', 'python', 'cros',
    102       'jetstream_services')),
    103     (cros_repair.ServoInstallRepair,
    104      'usb',
    105      (),
    106      ('ssh', 'writable', 'tpm', 'good_au', 'ext4', 'power', 'rwfw', 'python',
    107       'cros', 'jetstream_services')),
    108 )
    109 
    110 class CrosRepairUnittests(unittest.TestCase):
    111 
    112     maxDiff = None
    113 
    114     def test_cros_repair(self):
    115         verify_dag = cros_repair._cros_verify_dag()
    116         self.assertTupleEqual(verify_dag, CROS_VERIFY_DAG)
    117         self.check_verify_dag(verify_dag)
    118         repair_actions = cros_repair._cros_repair_actions()
    119         self.assertTupleEqual(repair_actions, CROS_REPAIR_ACTIONS)
    120         self.check_repair_actions(verify_dag, repair_actions)
    121 
    122     def test_moblab_repair(self):
    123         verify_dag = cros_repair._moblab_verify_dag()
    124         self.assertTupleEqual(verify_dag, MOBLAB_VERIFY_DAG)
    125         self.check_verify_dag(verify_dag)
    126         repair_actions = cros_repair._moblab_repair_actions()
    127         self.assertTupleEqual(repair_actions, MOBLAB_REPAIR_ACTIONS)
    128         self.check_repair_actions(verify_dag, repair_actions)
    129 
    130     def test_jetstream_repair(self):
    131         verify_dag = cros_repair._jetstream_verify_dag()
    132         self.assertTupleEqual(verify_dag, JETSTREAM_VERIFY_DAG)
    133         self.check_verify_dag(verify_dag)
    134         repair_actions = cros_repair._jetstream_repair_actions()
    135         self.assertTupleEqual(repair_actions, JETSTREAM_REPAIR_ACTIONS)
    136         self.check_repair_actions(verify_dag, repair_actions)
    137 
    138     def check_verify_dag(self, verify_dag):
    139         """Checks that dependency labels are defined."""
    140         labels = [n[1] for n in verify_dag]
    141         for node in verify_dag:
    142             for dep in node[2]:
    143                 self.assertIn(dep, labels)
    144 
    145     def check_repair_actions(self, verify_dag, repair_actions):
    146         """Checks that dependency and trigger labels are defined."""
    147         verify_labels = [n[1] for n in verify_dag]
    148         for action in repair_actions:
    149             deps = action[2]
    150             triggers = action[3]
    151             for label in deps + triggers:
    152                 self.assertIn(label, verify_labels)
    153 
    154 
    155 if __name__ == '__main__':
    156     unittest.main()
    157