# 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()