| # Copyright (c) 2017 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unit tests for server/cros/network/rf_switch/rf_switch.py.""" |
| |
| import unittest |
| |
| from autotest_lib.client.common_lib.test_utils import mock |
| from autotest_lib.server.cros.network.rf_switch import rf_mocks |
| from autotest_lib.server.cros.network.rf_switch import rf_switch |
| from autotest_lib.server.cros.network.rf_switch import scpi |
| |
| |
| class RfSwitchTestCases(unittest.TestCase): |
| """Unit tests for RfSwitch Methods.""" |
| |
| _HOST = '1.2.3.4' |
| |
| def setUp(self): |
| self.god = mock.mock_god(debug=False, fail_fast=True) |
| self.mock_rf_switch = rf_mocks.RfSwitchMock(self._HOST) |
| self.code = 0 |
| self.msg = '' |
| |
| def tearDown(self): |
| self.god.unstub_all() |
| |
| def _populate_stack_for_cmd(self, cmd): |
| """Helper to add call stacks to verify in mock. |
| |
| @param cmd: command to verify |
| """ |
| # monitor send for correct command. |
| self.god.stub_function(self.mock_rf_switch.socket, 'send') |
| self.mock_rf_switch.socket.send.expect_call('%s' % cmd) |
| self.mock_rf_switch.socket.send.expect_call( |
| '%s\n' % scpi.Scpi.CMD_ERROR_CHECK) |
| self.mock_rf_switch.socket.send.expect_call( |
| '%s\n' % rf_switch.RfSwitch._CMD_WAIT) |
| |
| self.god.stub_function_to_return( |
| self.mock_rf_switch.socket, 'recv', '%d, "%s"' % ( |
| self.code, self.msg)) |
| |
| def _populate_stack_for_query_commands(self, cmd, response): |
| """Helper to add call stacks for query commands |
| |
| @param cmd: command to verify |
| @param response: response to mock return |
| """ |
| self.god.stub_function(self.mock_rf_switch.socket, 'send') |
| self.mock_rf_switch.socket.send.expect_call('%s' % cmd) |
| |
| self.god.stub_function_to_return( |
| self.mock_rf_switch.socket, 'recv', response) |
| |
| def test_send_cmd_check_error(self): |
| """Verify send_cmd_check_error.""" |
| test_command = 'This is a command\n' |
| self._populate_stack_for_cmd(test_command) |
| self.mock_rf_switch.send_cmd_check_error(test_command) |
| self.god.check_playback() |
| |
| def test_send_cmd_check_error_throws_error(self): |
| """Verify send_cmd_check_error throws error.""" |
| test_command = 'This is a command' |
| code = 101 |
| msg = 'Error Message' |
| |
| self.god.stub_function_to_return( |
| self.mock_rf_switch.socket, 'recv', '%d, "%s"' % (code, msg)) |
| with self.assertRaises(rf_switch.RfSwitchException): |
| self.mock_rf_switch.send_cmd_check_error(test_command) |
| |
| def test_close_relays(self): |
| """Verify close_relays.""" |
| relays = 'R1:R3' |
| self._populate_stack_for_cmd( |
| '%s (@%s)\n' % (rf_switch.RfSwitch._CMD_CLOSE_RELAYS, relays)) |
| self.mock_rf_switch.close_relays(relays) |
| self.god.check_playback() |
| |
| def test_relays_closed(self): |
| """Verify relays_closed.""" |
| relays = 'R1:R3' |
| status = '1,0,1' |
| cmd = '%s (@%s)\n' % ( |
| rf_switch.RfSwitch._CMD_CHECK_RELAYS_CLOSED, relays) |
| self._populate_stack_for_query_commands(cmd, status) |
| self.mock_rf_switch.relays_closed(relays) |
| self.god.check_playback() |
| |
| def test_open_relays(self): |
| """Verify open_relays.""" |
| relays = 'R1:R3' |
| self._populate_stack_for_cmd( |
| '%s (@%s)\n' % (rf_switch.RfSwitch._CMD_OPEN_RELAYS, relays)) |
| self.mock_rf_switch.open_relays(relays) |
| self.god.check_playback() |
| |
| def test_open_all_relays(self): |
| """Verify open_all_relays.""" |
| self._populate_stack_for_cmd( |
| '%s\n' % rf_switch.RfSwitch._CMD_OPEN_ALL_RELAYS) |
| self.mock_rf_switch.open_all_relays() |
| self.god.check_playback() |
| |
| def test_set_verify_on(self): |
| """Verify set_verify for on.""" |
| relays = 'R1:R3' |
| status = True |
| self._populate_stack_for_cmd( |
| '%s 1,(@%s)\n' % (rf_switch.RfSwitch._CMD_SET_VERIFY, relays)) |
| self.mock_rf_switch.set_verify(relays, status) |
| self.god.check_playback() |
| |
| def test_set_verify_off(self): |
| """Verify set_verify for off.""" |
| relays = 'R1:R3' |
| status = False |
| self._populate_stack_for_cmd( |
| '%s 0,(@%s)\n' % (rf_switch.RfSwitch._CMD_SET_VERIFY, relays)) |
| self.mock_rf_switch.set_verify(relays, status) |
| self.god.check_playback() |
| |
| def test_get_verify(self): |
| """Verify get_verify.""" |
| relays = 'R1:R3' |
| status = '1,0,1' |
| cmd = '%s (@%s)\n' % (rf_switch.RfSwitch._CMD_GET_VERIFY, relays) |
| self._populate_stack_for_query_commands(cmd, status) |
| self.mock_rf_switch.get_verify(relays) |
| self.god.check_playback() |
| |
| def test_set_verify_inverted_true(self): |
| """Verify set_verify_inverted.""" |
| relays = 'R1:R3' |
| status = True |
| cmd = '%s INV,(@%s)\n' % ( |
| rf_switch.RfSwitch._CMD_SET_VERIFY_INVERTED, relays) |
| self._populate_stack_for_cmd(cmd) |
| self.mock_rf_switch.set_verify_inverted(relays, status) |
| self.god.check_playback() |
| |
| def test_set_verify_inverted_false(self): |
| """Verify set_verify_inverted.""" |
| relays = 'R1:R3' |
| status = False |
| cmd = '%s NORM,(@%s)\n' % ( |
| rf_switch.RfSwitch._CMD_SET_VERIFY_INVERTED, relays) |
| self._populate_stack_for_cmd(cmd) |
| self.mock_rf_switch.set_verify_inverted(relays, status) |
| self.god.check_playback() |
| |
| def test_get_verify_inverted(self): |
| """Verify get_verify_inverted.""" |
| relays = 'R1:R3' |
| status = '1,0,1' |
| cmd = '%s (@%s)\n' % ( |
| rf_switch.RfSwitch._CMD_GET_VERIFY_INVERTED, relays) |
| self._populate_stack_for_query_commands(cmd, status) |
| self.mock_rf_switch.get_verify_inverted(relays) |
| self.god.check_playback() |
| |
| def test_get_verify_state(self): |
| """Verify get_verify_state.""" |
| relays = 'R1:R3' |
| status = '1,0,1' |
| cmd = '%s (@%s)\n' % (rf_switch.RfSwitch._CMD_GET_VERIFY_STATE, relays) |
| self._populate_stack_for_query_commands(cmd, status) |
| self.mock_rf_switch.get_verify_state(relays) |
| self.god.check_playback() |
| |
| def test_busy(self): |
| """Verify busy.""" |
| status = '1' |
| cmd = '%s\n' % rf_switch.RfSwitch._CMD_CHECK_BUSY |
| self._populate_stack_for_query_commands(cmd, status) |
| busy = self.mock_rf_switch.busy |
| self.god.check_playback() |
| self.assertTrue(busy) |
| |
| def test_not_busy(self): |
| """Verify busy.""" |
| status = '0' |
| cmd = '%s\n' % rf_switch.RfSwitch._CMD_CHECK_BUSY |
| self._populate_stack_for_query_commands(cmd, status) |
| busy = self.mock_rf_switch.busy |
| self.god.check_playback() |
| self.assertFalse(busy) |
| |
| def test_wait(self): |
| """Verify wait.""" |
| self.god.stub_function(self.mock_rf_switch.socket, 'send') |
| self.mock_rf_switch.socket.send.expect_call( |
| '%s\n' % rf_switch.RfSwitch._CMD_WAIT) |
| self.mock_rf_switch.wait() |
| self.god.check_playback() |
| |
| def test_get_attenuation(self): |
| """Verify get_attenuation.""" |
| ap = 1 |
| attenuation = 100 |
| relays = rf_switch.RfSwitch._AP_ATTENUATOR_RELAYS_SHORT[ap] |
| return_relays = '1,1,0,1,1,0,0' |
| cmd = '%s (@%s)\n' % (rf_switch.RfSwitch._CMD_CHECK_RELAYS_CLOSED, |
| relays) |
| self._populate_stack_for_query_commands(cmd, return_relays) |
| response = self.mock_rf_switch.get_attenuation(ap) |
| self.god.check_playback() |
| self.assertEquals(response, attenuation, |
| 'get_attenuation returned %s, did not match %s' % ( |
| response, attenuation)) |
| |
| def test_set_attenuation(self): |
| """Verify set_attenuation.""" |
| ap = 1 |
| ap_relays = 'k1_49,k1_50,k1_51,k1_52,k1_53,k1_54,R1_9' |
| attenuation = 100 |
| relays_for_attenuation = 'k1_53,k1_52,k1_50,k1_49' |
| |
| self._populate_stack_for_cmd( |
| '%s (@%s)\n' % (rf_switch.RfSwitch._CMD_OPEN_RELAYS, ap_relays)) |
| self._populate_stack_for_cmd('%s (@%s)\n' % ( |
| rf_switch.RfSwitch._CMD_CLOSE_RELAYS, relays_for_attenuation)) |
| self.mock_rf_switch.set_attenuation(ap, attenuation) |
| self.god.check_playback() |
| |
| def test_set_attenuation_all(self): |
| """Verify we can set same attenuation to all.""" |
| |
| # 0 should close all relays |
| for x in xrange(rf_switch.RfSwitch._MAX_ENCLOSURE): |
| relays = ','.join(rf_switch.RfSwitch._AP_ATTENUATOR_RELAYS[x + 1]) |
| reverse_relays = ','.join( |
| rf_switch.RfSwitch._AP_ATTENUATOR_RELAYS[x + 1][::-1]) |
| self._populate_stack_for_cmd( |
| '%s (@%s)\n' % (rf_switch.RfSwitch._CMD_OPEN_RELAYS, relays)) |
| self._populate_stack_for_cmd('%s (@%s)\n' % ( |
| rf_switch.RfSwitch._CMD_CLOSE_RELAYS, reverse_relays)) |
| self.mock_rf_switch.set_attenuation(0, 0) |
| |
| # 127 should open all (close none) |
| for x in xrange(rf_switch.RfSwitch._MAX_ENCLOSURE): |
| relays = ','.join(rf_switch.RfSwitch._AP_ATTENUATOR_RELAYS[x + 1]) |
| self._populate_stack_for_cmd( |
| '%s (@%s)\n' % (rf_switch.RfSwitch._CMD_OPEN_RELAYS, relays)) |
| self._populate_stack_for_cmd('%s (@)\n' % ( |
| rf_switch.RfSwitch._CMD_CLOSE_RELAYS)) |
| self.mock_rf_switch.set_attenuation(0, 127) |
| self.god.check_playback() |
| |
| def test_set_attenuation_raise_error(self): |
| """Verify set_attenuation will raise error.""" |
| with self.assertRaises(ValueError): |
| self.mock_rf_switch.set_attenuation(5, 100) |
| with self.assertRaises(ValueError): |
| self.mock_rf_switch.set_attenuation(-1, 100) |
| |
| def test_connect_ap_client(self): |
| """Verify connect_ap_client sets correct relays.""" |
| relays = 'k1_1,k1_25' |
| ap = 1 |
| client = 1 |
| self._populate_stack_for_cmd( |
| '%s (@%s)\n' % (rf_switch.RfSwitch._CMD_CLOSE_RELAYS, relays)) |
| self.mock_rf_switch.connect_ap_client(ap, client) |
| self.god.check_playback() |
| |
| if __name__ == '__main__': |
| unittest.main() |
