[autotest] Add POE support to RPM server
Add POE support to RPM server. This allows us to hard reset
beaglebones utilizing the RPM server.
The core implementation resides in POEController in rpm_controller.py
To find out to which switch a servo is connected to,
we store the mapping information(servo->switch) in a csv-formatted file.
The file name is specified in rpm_config.ini.
rpm_dispatcher is modified to route poe requests to POEController instances.
Once rpm_dispatcher is initialized, it loads the mapping file to a dictionary
and uses it to determine the switch hostname and interface for a servo.
Simple integration tests are added in rpm_controller.py
Unit tests are added in rpm_controller_unittest.py
utils_unittest.py is added to test util functions in utils.py
servo-interface-mapping.csv should be updated if configuration
in switches changes.
BUG=chromium:232614
TEST=ran rpm_controller_unittest.py, rpm_dispacher_unittest.py,
frontend_server_unittest.py, utils_unittest.py, integration tests
in rpm_controller.py, test_client.py, and ran frontend_server.py,
rpm_dispatcher, rpm_client locally.
Change-Id: I294e34ad7efc6d9cc7543e1c57694466086d1791
Reviewed-on: https://gerrit.chromium.org/gerrit/55838
Reviewed-by: Simran Basi <sbasi@chromium.org>
Commit-Queue: Fang Deng <fdeng@chromium.org>
Tested-by: Fang Deng <fdeng@chromium.org>
diff --git a/site_utils/rpm_control_system/rpm_controller_unittest.py b/site_utils/rpm_control_system/rpm_controller_unittest.py
index d49e025..5b68078 100755
--- a/site_utils/rpm_control_system/rpm_controller_unittest.py
+++ b/site_utils/rpm_control_system/rpm_controller_unittest.py
@@ -3,10 +3,8 @@
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
-import logging
import mox
import pexpect
-import time
import unittest
import dli
@@ -37,6 +35,7 @@
self.ssh.sendline('%s %s' % (new_state, dut_hostname))
self.ssh.expect('Command successful', timeout=60)
self.ssh.sendline('logout')
+ self.ssh.close(force=True)
self.mox.ReplayAll()
self.assertTrue(self.rpm.queue_request(dut_hostname, new_state))
self.mox.VerifyAll()
@@ -55,6 +54,7 @@
self.ssh.expect('Command successful',
timeout=60).AndRaise(pexpect.TIMEOUT('Timed Out'))
self.ssh.sendline('logout')
+ self.ssh.close(force=True)
self.mox.ReplayAll()
self.assertFalse(self.rpm.queue_request(dut_hostname, new_state))
self.mox.VerifyAll()
@@ -78,7 +78,7 @@
def testSuccessfullyChangeOutlet(self):
"""Should return True if change was successful."""
- test_status_list_final = [[8,'chromeos-rack8a-host8','u\'OFF\'']]
+ test_status_list_final = [[8, 'chromeos-rack8a-host8','u\'OFF\'']]
self.dli_ps.statuslist().AndReturn(self.test_status_list_initial)
self.dli_ps.off(8)
self.dli_ps.statuslist().AndReturn(test_status_list_final)
@@ -90,7 +90,7 @@
def testUnsuccessfullyChangeOutlet(self):
"""Should return False if Outlet State does not change."""
- test_status_list_final = [[8,'chromeos-rack8a-host8','u\'ON\'']]
+ test_status_list_final = [[8, 'chromeos-rack8a-host8','u\'ON\'']]
self.dli_ps.statuslist().AndReturn(self.test_status_list_initial)
self.dli_ps.off(8)
self.dli_ps.statuslist().AndReturn(test_status_list_final)
@@ -109,5 +109,148 @@
self.mox.VerifyAll()
+class TestCiscoPOEController(mox.MoxTestBase):
+ """Test CiscoPOEController."""
+
+
+ STREAM_WELCOME = 'This is a POE switch.\n\nUser Name:'
+ STREAM_PWD = 'Password:'
+ STREAM_DEVICE = '\nchromeos2-poe-sw8#'
+ STREAM_CONFIG = 'chromeos2-poe-sw8(config)#'
+ STREAM_CONFIG_IF = 'chromeos2-poe-sw8(config-if)#'
+ STREAM_STATUS = ('\n '
+ 'Flow Link Back Mdix\n'
+ 'Port Type Duplex Speed Neg '
+ 'ctrl State Pressure Mode\n'
+ '-------- ------------ ------ ----- -------- '
+ '---- ----------- -------- -------\n'
+ 'fa32 100M-Copper Full 100 Enabled '
+ 'Off Up Disabled Off\n')
+ SERVO = 'chromeos1-rack3-host12-servo'
+ SWITCH = 'chromeos2-poe-switch8'
+ PORT = 'fa32'
+
+
+ def setUp(self):
+ super(TestCiscoPOEController, self).setUp()
+ self.mox.StubOutWithMock(pexpect.spawn, '_spawn')
+ self.mox.StubOutWithMock(pexpect.spawn, 'read_nonblocking')
+ self.mox.StubOutWithMock(pexpect.spawn, 'sendline')
+ servo_interface = {self.SERVO:(self.SWITCH, self.PORT)}
+ self.poe = rpm_controller.CiscoPOEController(
+ self.SWITCH, servo_interface)
+ pexpect.spawn._spawn(mox.IgnoreArg(), mox.IgnoreArg())
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_WELCOME)
+ pexpect.spawn.sendline(self.poe._username)
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_PWD)
+ pexpect.spawn.sendline(self.poe._password)
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_DEVICE)
+
+
+ def testLogin(self):
+ """Test we can log into the switch."""
+ self.mox.ReplayAll()
+ self.assertNotEqual(self.poe._login(), None)
+ self.mox.VerifyAll()
+
+
+ def _EnterConfigurationHelper(self, success=True):
+ """A helper function for testing entering configuration terminal.
+
+ @param success: True if we want the process to pass, False if we
+ want it to fail.
+ """
+ pexpect.spawn.sendline('configure terminal')
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_CONFIG)
+ pexpect.spawn.sendline('interface %s' % self.PORT)
+ if success:
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(),
+ mox.IgnoreArg()).AndReturn(self.STREAM_CONFIG_IF)
+ else:
+ self.mox.StubOutWithMock(pexpect.spawn, '__str__')
+ exception = pexpect.TIMEOUT(
+ 'Could not enter configuration terminal.')
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(),
+ mox.IgnoreArg()).MultipleTimes().AndRaise(exception)
+ pexpect.spawn.__str__().AndReturn('A pexpect.spawn object.')
+ pexpect.spawn.sendline('end')
+
+
+ def testSuccessfullyChangeOutlet(self):
+ """Should return True if change was successful."""
+ self._EnterConfigurationHelper()
+ pexpect.spawn.sendline('power inline auto')
+ pexpect.spawn.sendline('end')
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_DEVICE)
+ pexpect.spawn.sendline('show interface status %s' % self.PORT)
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_STATUS)
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(self.STREAM_DEVICE)
+ pexpect.spawn.sendline('exit')
+ self.mox.ReplayAll()
+ self.assertTrue(self.poe.queue_request(self.SERVO, 'ON'))
+ self.mox.VerifyAll()
+
+
+ def testUnableToEnterConfigurationTerminal(self):
+ """Should return False if unable to enter configuration terminal."""
+ self._EnterConfigurationHelper(success=False)
+ pexpect.spawn.sendline('exit')
+ self.mox.ReplayAll()
+ self.assertFalse(self.poe.queue_request(self.SERVO, 'ON'))
+ self.mox.VerifyAll()
+
+
+ def testUnableToExitConfigurationTerminal(self):
+ """Should return False if unable to exit configuration terminal."""
+ self.mox.StubOutWithMock(pexpect.spawn, '__str__')
+ self.mox.StubOutWithMock(rpm_controller.CiscoPOEController,
+ '_enter_configuration_terminal')
+ self.poe._enter_configuration_terminal(
+ mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(True)
+ pexpect.spawn.sendline('power inline auto')
+ pexpect.spawn.sendline('end')
+ exception = pexpect.TIMEOUT('Could not exit configuration terminal.')
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(),
+ mox.IgnoreArg()).MultipleTimes().AndRaise(exception)
+ pexpect.spawn.__str__().AndReturn('A pexpect.spawn object.')
+ pexpect.spawn.sendline('exit')
+ self.mox.ReplayAll()
+ self.assertFalse(self.poe.queue_request(self.SERVO, 'ON'))
+ self.mox.VerifyAll()
+
+
+ def testUnableToVerifyState(self):
+ """Should return False if unable to verify current state."""
+ self.mox.StubOutWithMock(pexpect.spawn, '__str__')
+ self.mox.StubOutWithMock(rpm_controller.CiscoPOEController,
+ '_enter_configuration_terminal')
+ self.mox.StubOutWithMock(rpm_controller.CiscoPOEController,
+ '_exit_configuration_terminal')
+ self.poe._enter_configuration_terminal(
+ mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(True)
+ pexpect.spawn.sendline('power inline auto')
+ self.poe._exit_configuration_terminal(mox.IgnoreArg()).AndReturn(True)
+ pexpect.spawn.sendline('show interface status %s' % self.PORT)
+ exception = pexpect.TIMEOUT('Could not verify state.')
+ pexpect.spawn.read_nonblocking(
+ mox.IgnoreArg(),
+ mox.IgnoreArg()).MultipleTimes().AndRaise(exception)
+ pexpect.spawn.__str__().AndReturn('A pexpect.spawn object.')
+ pexpect.spawn.sendline('exit')
+ self.mox.ReplayAll()
+ self.assertFalse(self.poe.queue_request(self.SERVO, 'ON'))
+ self.mox.VerifyAll()
+
+
if __name__ == "__main__":
unittest.main()