Home | History | Annotate | Download | only in rpm_control_system 
      1 #!/usr/bin/python
      2 # Copyright (c) 2012 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 mox
      7 import pexpect
      8 import unittest
      9 
     10 import dli
     11 
     12 import rpm_controller
     13 
     14 import common
     15 from autotest_lib.site_utils.rpm_control_system import utils
     16 
     17 
     18 class TestRPMControllerQueue(mox.MoxTestBase):
     19     """Test request can be queued and processed in controller.
     20     """
     21 
     22     def setUp(self):
     23         super(TestRPMControllerQueue, self).setUp()
     24         self.rpm = rpm_controller.SentryRPMController('chromeos-rack1-host8')
     25         self.powerunit_info = utils.PowerUnitInfo(
     26                 device_hostname='chromos-rack1-host8',
     27                 powerunit_hostname='chromeos-rack1-rpm1',
     28                 powerunit_type=utils.PowerUnitInfo.POWERUNIT_TYPES.RPM,
     29                 outlet='.A100',
     30                 hydra_hostname=None)
     31 
     32 
     33     def testQueueRequest(self):
     34         """Should create a new process to handle request."""
     35         new_state = 'ON'
     36         process = self.mox.CreateMockAnything()
     37         rpm_controller.multiprocessing.Process = self.mox.CreateMockAnything()
     38         rpm_controller.multiprocessing.Process(target=mox.IgnoreArg(),
     39                 args=mox.IgnoreArg()).AndReturn(process)
     40         process.start()
     41         process.join()
     42         self.mox.ReplayAll()
     43         self.assertFalse(self.rpm.queue_request(self.powerunit_info, new_state))
     44         self.mox.VerifyAll()
     45 
     46 
     47 class TestSentryRPMController(mox.MoxTestBase):
     48     """Test SentryRPMController."""
     49 
     50 
     51     def setUp(self):
     52         super(TestSentryRPMController, self).setUp()
     53         self.ssh = self.mox.CreateMockAnything()
     54         rpm_controller.pexpect.spawn = self.mox.CreateMockAnything()
     55         rpm_controller.pexpect.spawn(mox.IgnoreArg()).AndReturn(self.ssh)
     56         self.rpm = rpm_controller.SentryRPMController('chromeos-rack1-host8')
     57         self.powerunit_info = utils.PowerUnitInfo(
     58                 device_hostname='chromos-rack1-host8',
     59                 powerunit_hostname='chromeos-rack1-rpm1',
     60                 powerunit_type=utils.PowerUnitInfo.POWERUNIT_TYPES.RPM,
     61                 outlet='.A100',
     62                 hydra_hostname=None)
     63 
     64 
     65     def testSuccessfullyChangeOutlet(self):
     66         """Should return True if change was successful."""
     67         prompt = 'Switched CDU:'
     68         password = 'admn'
     69         new_state = 'ON'
     70         self.ssh.expect('Password:', timeout=60)
     71         self.ssh.sendline(password)
     72         self.ssh.expect(prompt, timeout=60)
     73         self.ssh.sendline('%s %s' % (new_state, self.powerunit_info.outlet))
     74         self.ssh.expect('Command successful', timeout=60)
     75         self.ssh.sendline('logout')
     76         self.ssh.close(force=True)
     77         self.mox.ReplayAll()
     78         self.assertTrue(self.rpm.set_power_state(
     79                 self.powerunit_info, new_state))
     80         self.mox.VerifyAll()
     81 
     82 
     83     def testUnsuccessfullyChangeOutlet(self):
     84         """Should return False if change was unsuccessful."""
     85         prompt = 'Switched CDU:'
     86         password = 'admn'
     87         new_state = 'ON'
     88         self.ssh.expect('Password:', timeout=60)
     89         self.ssh.sendline(password)
     90         self.ssh.expect(prompt, timeout=60)
     91         self.ssh.sendline('%s %s' % (new_state, self.powerunit_info.outlet))
     92         self.ssh.expect('Command successful',
     93                         timeout=60).AndRaise(pexpect.TIMEOUT('Timed Out'))
     94         self.ssh.sendline('logout')
     95         self.ssh.close(force=True)
     96         self.mox.ReplayAll()
     97         self.assertFalse(self.rpm.set_power_state(self.powerunit_info, new_state))
     98         self.mox.VerifyAll()
     99 
    100 
    101 class TestWebPoweredRPMController(mox.MoxTestBase):
    102     """Test WebPoweredRPMController."""
    103 
    104 
    105     def setUp(self):
    106         super(TestWebPoweredRPMController, self).setUp()
    107         self.dli_ps = self.mox.CreateMock(dli.powerswitch)
    108         hostname = 'chromeos-rack8a-rpm1'
    109         self.web_rpm = rpm_controller.WebPoweredRPMController(hostname,
    110                                                               self.dli_ps)
    111         outlet = 8
    112         dut = 'chromeos-rack8a-host8'
    113         # Outlet statuses are in the format "u'ON'"
    114         initial_state = 'u\'ON\''
    115         self.test_status_list_initial = [[outlet, dut, initial_state]]
    116         self.powerunit_info = utils.PowerUnitInfo(
    117                 device_hostname=dut,
    118                 powerunit_hostname=hostname,
    119                 powerunit_type=utils.PowerUnitInfo.POWERUNIT_TYPES.RPM,
    120                 outlet=outlet,
    121                 hydra_hostname=None)
    122 
    123 
    124     def testSuccessfullyChangeOutlet(self):
    125         """Should return True if change was successful."""
    126         test_status_list_final = [[8, 'chromeos-rack8a-host8','u\'OFF\'']]
    127         self.dli_ps.statuslist().AndReturn(self.test_status_list_initial)
    128         self.dli_ps.off(8)
    129         self.dli_ps.statuslist().AndReturn(test_status_list_final)
    130         self.mox.ReplayAll()
    131         self.assertTrue(self.web_rpm.set_power_state(
    132                 self.powerunit_info, 'OFF'))
    133         self.mox.VerifyAll()
    134 
    135 
    136     def testUnsuccessfullyChangeOutlet(self):
    137         """Should return False if Outlet State does not change."""
    138         test_status_list_final = [[8, 'chromeos-rack8a-host8','u\'ON\'']]
    139         self.dli_ps.statuslist().AndReturn(self.test_status_list_initial)
    140         self.dli_ps.off(8)
    141         self.dli_ps.statuslist().AndReturn(test_status_list_final)
    142         self.mox.ReplayAll()
    143         self.assertFalse(self.web_rpm.set_power_state(
    144                 self.powerunit_info, 'OFF'))
    145         self.mox.VerifyAll()
    146 
    147 
    148     def testNoOutlet(self):
    149         """Should return False if DUT hostname is not on the RPM device."""
    150         self.powerunit_info.outlet=None
    151         self.assertFalse(self.web_rpm.set_power_state(
    152                 self.powerunit_info, 'OFF'))
    153 
    154 
    155 class TestCiscoPOEController(mox.MoxTestBase):
    156     """Test CiscoPOEController."""
    157 
    158 
    159     STREAM_WELCOME = 'This is a POE switch.\n\nUser Name:'
    160     STREAM_PWD = 'Password:'
    161     STREAM_DEVICE = '\nchromeos2-poe-sw8#'
    162     STREAM_CONFIG = 'chromeos2-poe-sw8(config)#'
    163     STREAM_CONFIG_IF = 'chromeos2-poe-sw8(config-if)#'
    164     STREAM_STATUS = ('\n                                             '
    165                      'Flow Link          Back   Mdix\n'
    166                      'Port     Type         Duplex  Speed Neg      '
    167                      'ctrl State       Pressure Mode\n'
    168                      '-------- ------------ ------  ----- -------- '
    169                      '---- ----------- -------- -------\n'
    170                      'fa32     100M-Copper  Full    100   Enabled  '
    171                      'Off  Up          Disabled Off\n')
    172     SERVO = 'chromeos1-rack3-host12-servo'
    173     SWITCH = 'chromeos2-poe-switch8'
    174     PORT = 'fa32'
    175     POWERUNIT_INFO = utils.PowerUnitInfo(
    176             device_hostname=PORT,
    177             powerunit_hostname=SERVO,
    178             powerunit_type=utils.PowerUnitInfo.POWERUNIT_TYPES.POE,
    179             outlet=PORT,
    180             hydra_hostname=None)
    181 
    182 
    183     def setUp(self):
    184         super(TestCiscoPOEController, self).setUp()
    185         self.mox.StubOutWithMock(pexpect.spawn, '_spawn')
    186         self.mox.StubOutWithMock(pexpect.spawn, 'read_nonblocking')
    187         self.mox.StubOutWithMock(pexpect.spawn, 'sendline')
    188         self.poe = rpm_controller.CiscoPOEController(self.SWITCH)
    189         pexpect.spawn._spawn(mox.IgnoreArg(), mox.IgnoreArg())
    190         pexpect.spawn.read_nonblocking(
    191                 mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_WELCOME)
    192         pexpect.spawn.sendline(self.poe._username)
    193         pexpect.spawn.read_nonblocking(
    194                 mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_PWD)
    195         pexpect.spawn.sendline(self.poe._password)
    196         pexpect.spawn.read_nonblocking(
    197                 mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_DEVICE)
    198 
    199 
    200     def testLogin(self):
    201         """Test we can log into the switch."""
    202         self.mox.ReplayAll()
    203         self.assertNotEqual(self.poe._login(), None)
    204         self.mox.VerifyAll()
    205 
    206 
    207     def _EnterConfigurationHelper(self, success=True):
    208         """A helper function for testing entering configuration terminal.
    209 
    210         @param success: True if we want the process to pass, False if we
    211                         want it to fail.
    212         """
    213         pexpect.spawn.sendline('configure terminal')
    214         pexpect.spawn.read_nonblocking(
    215                 mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_CONFIG)
    216         pexpect.spawn.sendline('interface %s' % self.PORT)
    217         if success:
    218             pexpect.spawn.read_nonblocking(
    219                     mox.IgnoreArg(),
    220                     mox.IgnoreArg()).AndReturn(self.STREAM_CONFIG_IF)
    221         else:
    222             self.mox.StubOutWithMock(pexpect.spawn, '__str__')
    223             exception = pexpect.TIMEOUT(
    224                     'Could not enter configuration terminal.')
    225             pexpect.spawn.read_nonblocking(
    226                     mox.IgnoreArg(),
    227                     mox.IgnoreArg()).MultipleTimes().AndRaise(exception)
    228             pexpect.spawn.__str__().AndReturn('A pexpect.spawn object.')
    229             pexpect.spawn.sendline('end')
    230 
    231 
    232     def testSuccessfullyChangeOutlet(self):
    233         """Should return True if change was successful."""
    234         self._EnterConfigurationHelper()
    235         pexpect.spawn.sendline('power inline auto')
    236         pexpect.spawn.sendline('end')
    237         pexpect.spawn.read_nonblocking(
    238                 mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_DEVICE)
    239         pexpect.spawn.sendline('show interface status %s' % self.PORT)
    240         pexpect.spawn.read_nonblocking(
    241                 mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_STATUS)
    242         pexpect.spawn.read_nonblocking(
    243                 mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_DEVICE)
    244         pexpect.spawn.sendline('exit')
    245         self.mox.ReplayAll()
    246         self.assertTrue(self.poe.set_power_state(self.POWERUNIT_INFO, 'ON'))
    247         self.mox.VerifyAll()
    248 
    249 
    250     def testUnableToEnterConfigurationTerminal(self):
    251         """Should return False if unable to enter configuration terminal."""
    252         self._EnterConfigurationHelper(success=False)
    253         pexpect.spawn.sendline('exit')
    254         self.mox.ReplayAll()
    255         self.assertFalse(self.poe.set_power_state(self.POWERUNIT_INFO, 'ON'))
    256         self.mox.VerifyAll()
    257 
    258 
    259     def testUnableToExitConfigurationTerminal(self):
    260         """Should return False if unable to exit configuration terminal."""
    261         self.mox.StubOutWithMock(pexpect.spawn, '__str__')
    262         self.mox.StubOutWithMock(rpm_controller.CiscoPOEController,
    263                                  '_enter_configuration_terminal')
    264         self.poe._enter_configuration_terminal(
    265                 mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(True)
    266         pexpect.spawn.sendline('power inline auto')
    267         pexpect.spawn.sendline('end')
    268         exception = pexpect.TIMEOUT('Could not exit configuration terminal.')
    269         pexpect.spawn.read_nonblocking(
    270                 mox.IgnoreArg(),
    271                 mox.IgnoreArg()).MultipleTimes().AndRaise(exception)
    272         pexpect.spawn.__str__().AndReturn('A pexpect.spawn object.')
    273         pexpect.spawn.sendline('exit')
    274         self.mox.ReplayAll()
    275         self.assertFalse(self.poe.set_power_state(self.POWERUNIT_INFO, 'ON'))
    276         self.mox.VerifyAll()
    277 
    278 
    279     def testUnableToVerifyState(self):
    280         """Should return False if unable to verify current state."""
    281         self.mox.StubOutWithMock(pexpect.spawn, '__str__')
    282         self.mox.StubOutWithMock(rpm_controller.CiscoPOEController,
    283                                  '_enter_configuration_terminal')
    284         self.mox.StubOutWithMock(rpm_controller.CiscoPOEController,
    285                                  '_exit_configuration_terminal')
    286         self.poe._enter_configuration_terminal(
    287                 mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(True)
    288         pexpect.spawn.sendline('power inline auto')
    289         self.poe._exit_configuration_terminal(mox.IgnoreArg()).AndReturn(True)
    290         pexpect.spawn.sendline('show interface status %s' % self.PORT)
    291         exception = pexpect.TIMEOUT('Could not verify state.')
    292         pexpect.spawn.read_nonblocking(
    293                 mox.IgnoreArg(),
    294                 mox.IgnoreArg()).MultipleTimes().AndRaise(exception)
    295         pexpect.spawn.__str__().AndReturn('A pexpect.spawn object.')
    296         pexpect.spawn.sendline('exit')
    297         self.mox.ReplayAll()
    298         self.assertFalse(self.poe.set_power_state(self.POWERUNIT_INFO, 'ON'))
    299         self.mox.VerifyAll()
    300 
    301 
    302 if __name__ == "__main__":
    303     unittest.main()
    304