• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/python2
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    def setUp(self):
22        super(TestRPMControllerQueue, self).setUp()
23        self.rpm = rpm_controller.SentryRPMController('chromeos-rack1-host8')
24        self.powerunit_info = utils.PowerUnitInfo(
25                device_hostname='chromos-rack1-host8',
26                powerunit_hostname='chromeos-rack1-rpm1',
27                powerunit_type=utils.PowerUnitInfo.POWERUNIT_TYPES.RPM,
28                outlet='.A100',
29                hydra_hostname=None)
30
31
32    def testQueueRequest(self):
33        """Should create a new process to handle request."""
34        new_state = 'ON'
35        process = self.mox.CreateMockAnything()
36        rpm_controller.multiprocessing.Process = self.mox.CreateMockAnything()
37        rpm_controller.multiprocessing.Process(target=mox.IgnoreArg(),
38                args=mox.IgnoreArg()).AndReturn(process)
39        process.start()
40        process.join()
41        self.mox.ReplayAll()
42        self.assertFalse(self.rpm.queue_request(self.powerunit_info, new_state))
43        self.mox.VerifyAll()
44
45
46class TestSentryRPMController(mox.MoxTestBase):
47    """Test SentryRPMController."""
48
49
50    def setUp(self):
51        super(TestSentryRPMController, self).setUp()
52        self.ssh = self.mox.CreateMockAnything()
53        rpm_controller.pexpect.spawn = self.mox.CreateMockAnything()
54        rpm_controller.pexpect.spawn(mox.IgnoreArg()).AndReturn(self.ssh)
55        self.rpm = rpm_controller.SentryRPMController('chromeos-rack1-host8')
56        self.powerunit_info = utils.PowerUnitInfo(
57                device_hostname='chromos-rack1-host8',
58                powerunit_hostname='chromeos-rack1-rpm1',
59                powerunit_type=utils.PowerUnitInfo.POWERUNIT_TYPES.RPM,
60                outlet='.A100',
61                hydra_hostname=None)
62
63
64    def testSuccessfullyChangeOutlet(self):
65        """Should return True if change was successful."""
66        prompt = ['Switched CDU:', 'Switched PDU:']
67        password = 'admn'
68        new_state = 'ON'
69        self.ssh.expect('Password:', timeout=60)
70        self.ssh.sendline(password)
71        self.ssh.expect(prompt, timeout=60)
72        self.ssh.sendline('%s %s' % (new_state, self.powerunit_info.outlet))
73        self.ssh.expect('Command successful', timeout=60)
74        self.ssh.sendline('logout')
75        self.ssh.close(force=True)
76        self.mox.ReplayAll()
77        self.assertTrue(self.rpm.set_power_state(
78                self.powerunit_info, new_state))
79        self.mox.VerifyAll()
80
81
82    def testUnsuccessfullyChangeOutlet(self):
83        """Should return False if change was unsuccessful."""
84        prompt = ['Switched CDU:', 'Switched PDU:']
85        password = 'admn'
86        new_state = 'ON'
87        self.ssh.expect('Password:', timeout=60)
88        self.ssh.sendline(password)
89        self.ssh.expect(prompt, timeout=60)
90        self.ssh.sendline('%s %s' % (new_state, self.powerunit_info.outlet))
91        self.ssh.expect('Command successful',
92                        timeout=60).AndRaise(pexpect.TIMEOUT('Timed Out'))
93        self.ssh.sendline('logout')
94        self.ssh.close(force=True)
95        self.mox.ReplayAll()
96        self.assertFalse(
97            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.off(8)
128        self.dli_ps.statuslist().AndReturn(test_status_list_final)
129        self.mox.ReplayAll()
130        self.assertTrue(self.web_rpm.set_power_state(
131                self.powerunit_info, 'OFF'))
132        self.mox.VerifyAll()
133
134
135    def testUnsuccessfullyChangeOutlet(self):
136        """Should return False if Outlet State does not change."""
137        test_status_list_final = [[8, 'chromeos-rack8a-host8', 'u\'ON\'']]
138        self.dli_ps.off(8)
139        self.dli_ps.statuslist().AndReturn(test_status_list_final)
140        self.mox.ReplayAll()
141        self.assertFalse(self.web_rpm.set_power_state(
142                self.powerunit_info, 'OFF'))
143        self.mox.VerifyAll()
144
145
146    def testNoOutlet(self):
147        """Should return False if DUT hostname is not on the RPM device."""
148        self.powerunit_info.outlet=None
149        self.assertFalse(self.web_rpm.set_power_state(
150                self.powerunit_info, 'OFF'))
151
152
153class TestCiscoPOEController(mox.MoxTestBase):
154    """Test CiscoPOEController."""
155
156    DEVICE = 'chromeos2-poe-sw8#'
157    MATCHER = 'Port\\s+.*%s(\\s+(\\S+)){6,6}.*%s'
158    PORT = 'fa32'
159    PWD = 'Password:'
160    SERVO = 'chromeos1-rack3-host12-servo'
161    SWITCH = 'chromeos2-poe-switch8'
162    POWERUNIT_INFO = utils.PowerUnitInfo(
163            device_hostname=PORT,
164            powerunit_hostname=SERVO,
165            powerunit_type=utils.PowerUnitInfo.POWERUNIT_TYPES.POE,
166            outlet=PORT,
167            hydra_hostname=None)
168
169    def testLogin(self):
170        """Test we can log into the switch."""
171        rpm_controller.pexpect.spawn = self.mox.CreateMockAnything()
172        mock_ssh = self.mox.CreateMockAnything()
173        rpm_controller.pexpect.spawn(mox.IgnoreArg()).AndReturn(mock_ssh)
174        sut = rpm_controller.CiscoPOEController(self.SWITCH)
175        mock_ssh.expect(sut.POE_USERNAME_PROMPT, timeout=sut.LOGIN_TIMEOUT)
176        mock_ssh.sendline(mox.IgnoreArg())
177        mock_ssh.expect(self.PWD, timeout=sut.LOGIN_TIMEOUT)
178        mock_ssh.sendline(mox.IgnoreArg())
179        mock_ssh.expect(self.DEVICE, timeout=sut.LOGIN_TIMEOUT)
180        self.mox.ReplayAll()
181        self.assertIsNotNone(sut._login())
182        self.mox.VerifyAll()
183
184    def testSuccessfullyChangePowerState(self):
185        """Should return True if change was successful."""
186        sut = rpm_controller.CiscoPOEController(self.SWITCH)
187        mock_ssh = self.mox.CreateMockAnything()
188        self.mox.StubOutWithMock(sut, '_login')
189        sut._login().AndReturn(mock_ssh)
190        self.mox.StubOutWithMock(sut, '_verify_state')
191        sut._verify_state(self.PORT, 'ON', mock_ssh).AndReturn(True)
192        # _enter_configuration_terminal
193        mock_ssh.sendline(sut.CONFIG)
194        mock_ssh.expect(sut.config_prompt, timeout=sut.CMD_TIMEOUT)
195        mock_ssh.sendline(sut.CONFIG_IF % self.PORT)
196        mock_ssh.expect(sut.config_if_prompt, timeout=sut.CMD_TIMEOUT)
197        # _change_state
198        mock_ssh.sendline(sut.SET_STATE_ON)
199        # _exit_configuration_terminal
200        mock_ssh.sendline(sut.END_CMD)
201        mock_ssh.expect(sut.poe_prompt, timeout=sut.CMD_TIMEOUT)
202        # exit
203        mock_ssh.sendline(sut.EXIT_CMD)
204        mock_ssh.close(force=True)
205        self.mox.ReplayAll()
206        self.assertTrue(sut.set_power_state(self.POWERUNIT_INFO, 'ON'))
207        self.mox.VerifyAll()
208
209    def testUnableToEnterConfigurationTerminal(self):
210        """Should return False if unable to enter configuration terminal."""
211        exception = pexpect.TIMEOUT('Could not enter configuration terminal.')
212        sut = rpm_controller.CiscoPOEController(self.SWITCH)
213        timeout = sut.CMD_TIMEOUT
214        mock_ssh = self.mox.CreateMockAnything()
215        self.mox.StubOutWithMock(sut, '_login')
216        sut._login().AndReturn(mock_ssh)
217        mock_ssh.sendline(sut.CONFIG)
218        mock_ssh.expect(sut.config_prompt, timeout=timeout)
219        mock_ssh.sendline(sut.CONFIG_IF % self.PORT)
220        config_if_prompt = sut.config_if_prompt
221        mock_ssh.expect(config_if_prompt, timeout=timeout).AndRaise(exception)
222        mock_ssh.sendline(sut.END_CMD)
223        mock_ssh.sendline(sut.EXIT_CMD)
224        mock_ssh.close(force=True)
225        self.mox.ReplayAll()
226        self.assertFalse(sut.set_power_state(self.POWERUNIT_INFO, mock_ssh))
227        self.mox.VerifyAll()
228
229    def testUnableToExitConfigurationTerminal(self):
230        """Should return False if unable to exit configuration terminal."""
231        exception = pexpect.TIMEOUT('Could not exit configuration terminal.')
232        sut = rpm_controller.CiscoPOEController(self.SWITCH)
233        mock_ssh = self.mox.CreateMockAnything()
234        self.mox.StubOutWithMock(sut, '_login')
235        self.mox.StubOutWithMock(sut, '_enter_configuration_terminal')
236        sut._login().AndReturn(mock_ssh)
237        sut._enter_configuration_terminal(self.PORT, mock_ssh).AndReturn(True)
238        mock_ssh.sendline(sut.SET_STATE_ON)
239        mock_ssh.sendline(sut.END_CMD)
240        mock_ssh.expect(
241                self.DEVICE, timeout=sut.CMD_TIMEOUT).AndRaise(exception)
242        mock_ssh.sendline(sut.EXIT_CMD)
243        mock_ssh.close(force=True)
244        self.mox.ReplayAll()
245        self.assertFalse(sut.set_power_state(self.POWERUNIT_INFO, 'ON'))
246        self.mox.VerifyAll()
247
248    def testUnableToVerifyState(self):
249        """Should return False if unable to verify current state."""
250        sut = rpm_controller.CiscoPOEController(self.SWITCH)
251        mock_ssh = self.mox.CreateMockAnything()
252        self.mox.StubOutWithMock(sut, '_login')
253        self.mox.StubOutWithMock(sut, '_enter_configuration_terminal')
254        self.mox.StubOutWithMock(sut, '_exit_configuration_terminal')
255        sut._login().AndReturn(mock_ssh)
256        sut._enter_configuration_terminal(self.PORT, mock_ssh).AndReturn(True)
257        sut._exit_configuration_terminal(mock_ssh).AndReturn(True)
258        mock_ssh.sendline(sut.SET_STATE_ON)
259        mock_ssh.sendline(sut.CHECK_INTERFACE_STATE % self.PORT)
260        exception = pexpect.TIMEOUT('Could not verify state.')
261        matcher = self.MATCHER % (self.PORT, self.DEVICE)
262        mock_ssh.expect(matcher, timeout=sut.CMD_TIMEOUT).AndRaise(exception)
263        mock_ssh.sendline(sut.EXIT_CMD)
264        mock_ssh.close(force=True)
265        self.mox.ReplayAll()
266        self.assertFalse(sut.set_power_state(self.POWERUNIT_INFO, 'ON'))
267        self.mox.VerifyAll()
268
269
270if __name__ == "__main__":
271    unittest.main()
272