• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
6import mox
7import pexpect
8import unittest
9
10import dli
11
12import rpm_controller
13
14import common
15from autotest_lib.site_utils.rpm_control_system import utils
16
17
18class 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
47class 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
101class 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
155class 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
302if __name__ == "__main__":
303    unittest.main()
304