build.tests.unit.test_main.TestMain.test_set_time() - Code Metrics - Inspection of "release 2022.3.1: fix: generate valid unicast src..." - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#88)

by Vinicius

created 2023-06-28 18:33 UTC

TestMain.test_set_time() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	11
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	1
dl	0
loc	11
ccs	7
cts	7
cp	1
crap	1
rs	10
c	0
b	0
f	0

"""Test Main methods."""
from unittest import TestCase
from unittest.mock import AsyncMock, MagicMock, call, patch

from kytos.lib.helpers import (get_controller_mock, get_kytos_event_mock,
                               get_switch_mock, get_test_client)

from kytos.core.events import KytosEvent
from napps.kytos.of_lldp.utils import get_cookie
from tests.helpers import get_topology_mock


@patch('kytos.core.controller.Controller.get_switch_by_dpid')

@patch('napps.kytos.of_lldp.main.Main._unpack_non_empty')
@patch('napps.kytos.of_lldp.main.UBInt32')
@patch('napps.kytos.of_lldp.main.DPID')
@patch('napps.kytos.of_lldp.main.LLDP')
@patch('napps.kytos.of_lldp.main.Ethernet')
async def test_on_ofpt_packet_in(*args):
    """Test on_ofpt_packet_in."""
    (mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
     mock_unpack_non_empty, mock_get_switch_by_dpid) = args

    # pylint: disable=bad-option-value, import-outside-toplevel
    from napps.kytos.of_lldp.main import Main
    Main.get_liveness_controller = MagicMock()
    topology = get_topology_mock()
    controller = get_controller_mock()
    controller.buffers.app.aput = AsyncMock()
    controller.switches = topology.switches
    napp = Main(controller)
    napp.loop_manager.process_if_looped = AsyncMock()
    napp.liveness_manager.consume_hello_if_enabled = AsyncMock()

    switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
    message = MagicMock(in_port=1, data='data')
    event = KytosEvent('ofpt_packet_in', content={'source': switch.connection,
                       'message': message})

    mocked, ethernet, lldp, dpid, port_b = [MagicMock() for _ in range(5)]
    mocked.value = 1
    mock_ubint32.return_value = mocked
    ethernet.ether_type = 0x88CC
    ethernet.data = 'eth_data'
    lldp.chassis_id.sub_value = 'chassis_id'
    lldp.port_id.sub_value = 'port_id'
    dpid.value = "00:00:00:00:00:00:00:02"
    port_b.value = 2

    mock_unpack_non_empty.side_effect = [ethernet, lldp, dpid, port_b]
    mock_get_switch_by_dpid.return_value = get_switch_mock(dpid.value,
                                                           0x04)
    await napp.on_ofpt_packet_in(event)

    calls = [call(mock_ethernet, message.data),
             call(mock_lldp, ethernet.data),
             call(mock_dpid, lldp.chassis_id.sub_value),
             call(mock_ubint32, lldp.port_id.sub_value)]
    mock_unpack_non_empty.assert_has_calls(calls)
    assert napp.loop_manager.process_if_looped.call_count == 1
    assert napp.liveness_manager.consume_hello_if_enabled.call_count == 1
    assert controller.buffers.app.aput.call_count == 1


@patch('kytos.core.controller.Controller.get_switch_by_dpid')

@patch('napps.kytos.of_lldp.main.Main._unpack_non_empty')
@patch('napps.kytos.of_lldp.main.UBInt32')
@patch('napps.kytos.of_lldp.main.DPID')
@patch('napps.kytos.of_lldp.main.LLDP')
@patch('napps.kytos.of_lldp.main.Ethernet')
async def test_on_ofpt_packet_in_early_intf(*args):
    """Test on_ofpt_packet_in early intf return."""
    (mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
     mock_unpack_non_empty, mock_get_switch_by_dpid) = args

    # pylint: disable=bad-option-value, import-outside-toplevel
    from napps.kytos.of_lldp.main import Main
    Main.get_liveness_controller = MagicMock()
    topology = get_topology_mock()
    controller = get_controller_mock()
    controller.buffers.app.aput = AsyncMock()
    controller.switches = topology.switches
    napp = Main(controller)
    napp.loop_manager.process_if_looped = AsyncMock()
    napp.liveness_manager.consume_hello_if_enabled = AsyncMock()

    switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
    message = MagicMock(in_port=1, data='data')
    event = KytosEvent('ofpt_packet_in', content={'source': switch.connection,
                       'message': message})

    mocked, ethernet, lldp, dpid, port_b = [MagicMock() for _ in range(5)]
    mocked.value = 1
    mock_ubint32.return_value = mocked
    ethernet.ether_type = 0x88CC
    ethernet.data = 'eth_data'
    lldp.chassis_id.sub_value = 'chassis_id'
    lldp.port_id.sub_value = 'port_id'
    dpid.value = "00:00:00:00:00:00:00:02"
    port_b.value = 2

    mock_unpack_non_empty.side_effect = [ethernet, lldp, dpid, port_b]
    mock_get_switch_by_dpid.return_value = get_switch_mock(dpid.value,
                                                           0x04)
    switch.get_interface_by_port_no = MagicMock(return_value=None)
    await napp.on_ofpt_packet_in(event)

    calls = [call(mock_ethernet, message.data),
             call(mock_lldp, ethernet.data),
             call(mock_dpid, lldp.chassis_id.sub_value),
             call(mock_ubint32, lldp.port_id.sub_value)]
    mock_unpack_non_empty.assert_has_calls(calls)
    switch.get_interface_by_port_no.assert_called()
    # early return shouldn't allow these to get called
    assert napp.loop_manager.process_if_looped.call_count == 0
    assert napp.liveness_manager.consume_hello_if_enabled.call_count == 0
    assert controller.buffers.app.aput.call_count == 0


# pylint: disable=protected-access,too-many-public-methods
class TestMain(TestCase):
    """Tests for the Main class."""

    def setUp(self):
        """Execute steps before each tests."""
        self.server_name_url = 'http://127.0.0.1:8181/api/kytos/of_lldp'

        patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
        # pylint: disable=bad-option-value, import-outside-toplevel
        from napps.kytos.of_lldp.main import Main
        self.addCleanup(patch.stopall)
        Main.get_liveness_controller = MagicMock()
        self.topology = get_topology_mock()
        controller = get_controller_mock()
        controller.switches = self.topology.switches

        self.napp = Main(controller)

    def get_topology_interfaces(self):
        """Return interfaces present in topology."""
        interfaces = []
        for switch in list(self.topology.switches.values()):
            interfaces += list(switch.interfaces.values())
        return interfaces

    @patch('napps.kytos.of_lldp.main.of_msg_prio')
    @patch('kytos.core.buffers.KytosEventBuffer.put')
    @patch('napps.kytos.of_lldp.main.KytosEvent')
    @patch('napps.kytos.of_lldp.main.VLAN')
    @patch('napps.kytos.of_lldp.main.Ethernet')
    @patch('napps.kytos.of_lldp.main.DPID')
    @patch('napps.kytos.of_lldp.main.LLDP')
    def test_execute(self, *args):
        """Test execute method."""
        (_, _, mock_ethernet, _, mock_kytos_event,
         mock_buffer_put, mock_of_msg_prio) = args

        ethernet = MagicMock()
        ethernet.pack.return_value = 'pack'
        interfaces = self.get_topology_interfaces()
        po_args = [(interface.switch.connection.protocol.version,
                    interface.port_number, 'pack') for interface in interfaces]

        mock_ethernet.return_value = ethernet
        mock_kytos_event.side_effect = po_args

        mock_publish_stopped = MagicMock()
        self.napp.try_to_publish_stopped_loops = mock_publish_stopped
        self.napp.execute()

        mock_of_msg_prio.assert_called()
        mock_buffer_put.assert_has_calls([call(arg)
                                          for arg in po_args])
        mock_publish_stopped.assert_called()

    @patch('requests.delete')
    @patch('requests.post')
    def test_handle_lldp_flows(self, mock_post, mock_delete):
        """Test handle_lldp_flow method."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        self.napp.controller.switches = {dpid: switch}
        event_post = get_kytos_event_mock(name='kytos/topology.switch.enabled',
                                          content={'dpid': dpid})

        event_del = get_kytos_event_mock(name='kytos/topology.switch.disabled',
                                         content={'dpid': dpid})

        mock_post.return_value = MagicMock(status_code=202)
        mock_delete.return_value = MagicMock(status_code=202)

        self.napp._handle_lldp_flows(event_post)
        mock_post.assert_called()

        self.napp._handle_lldp_flows(event_del)
        mock_delete.assert_called()

    @patch("time.sleep")
    @patch("requests.post")
    def test_handle_lldp_flows_retries(self, mock_post, _):
        """Test handle_lldp_flow method retries."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        self.napp.controller.switches = {dpid: switch}
        event_post = get_kytos_event_mock(name="kytos/topology.switch.enabled",
                                          content={"dpid": dpid})

        mock = MagicMock()
        mock.request.method = "POST"
        mock.status_code = 500
        mock.text = "some_err"
        mock_post.return_value = mock
        self.napp._handle_lldp_flows(event_post)
        self.assertTrue(mock_post.call_count, 3)

    @patch('napps.kytos.of_lldp.main.PO13')
    @patch('napps.kytos.of_lldp.main.AO13')
    def test_build_lldp_packet_out(self, *args):
        """Test _build_lldp_packet_out method."""
        (mock_ao13, mock_po13) = args

        ao13 = MagicMock()
        po13 = MagicMock()
        po13.actions = []

        mock_ao13.return_value = ao13
        mock_po13.return_value = po13

        packet_out13 = self.napp._build_lldp_packet_out(0x04, 2, 'data2')
        packet_out14 = self.napp._build_lldp_packet_out(0x05, 3, 'data3')

        self.assertEqual(packet_out13.data, 'data2')
        self.assertEqual(packet_out13.actions, [ao13])
        self.assertEqual(packet_out13.actions[0].port, 2)

        self.assertIsNone(packet_out14)

    @patch('napps.kytos.of_lldp.main.settings')
    @patch('napps.kytos.of_lldp.main.EtherType')
    @patch('napps.kytos.of_lldp.main.Port13')
    def test_build_lldp_flow(self, *args):
        """Test _build_lldp_flow method."""
        (mock_v0x04_port, mock_ethertype,
         mock_settings) = args
        self.napp.vlan_id = None
        mock_v0x04_port.OFPP_CONTROLLER = 1234

        mock_ethertype.LLDP = 10
        mock_settings.FLOW_VLAN_VID = None
        mock_settings.FLOW_PRIORITY = 1500
        mock_settings.TABLE_ID = 0
        dpid = "00:00:00:00:00:00:00:01"

        flow = {}
        match = {}
        flow['priority'] = 1500
        flow['table_id'] = 0
        match['dl_type'] = 10

        flow['match'] = match
        expected_flow_v0x04 = flow.copy()
        expected_flow_v0x04['cookie'] = get_cookie(dpid)
        expected_flow_v0x04['cookie_mask'] = 0xffffffffffffffff

        expected_flow_v0x04['actions'] = [{'action_type': 'output',
                                           'port': 1234}]

        flow_mod10 = self.napp._build_lldp_flow(0x01, get_cookie(dpid))
        flow_mod13 = self.napp._build_lldp_flow(0x04, get_cookie(dpid))

        self.assertIsNone(flow_mod10)
        self.assertDictEqual(flow_mod13, expected_flow_v0x04)

    def test_unpack_non_empty(self):
        """Test _unpack_non_empty method."""
        desired_class = MagicMock()
        data = MagicMock()
        data.value = 'data'

        obj = self.napp._unpack_non_empty(desired_class, data)

        obj.unpack.assert_called_with('data')

    def test_get_data(self):
        """Test _get_data method."""
        req = MagicMock()
        interfaces = ['00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:01:2']
        req.get_json.return_value = {'interfaces': interfaces}

        data = self.napp._get_data(req)

        self.assertEqual(data, interfaces)

    def test_load_liveness(self) -> None:
        """Test load_liveness."""
        self.napp.load_liveness()
        count = self.napp.liveness_controller.get_enabled_interfaces.call_count
        assert count == 1

    def test_handle_topology_loaded(self) -> None:
        """Test handle_topology_loaded."""
        event = KytosEvent("kytos/topology.topology_loaded",
                           content={"topology": {}})
        self.napp.load_liveness = MagicMock()
        self.napp.loop_manager.handle_topology_loaded = MagicMock()
        self.napp.handle_topology_loaded(event)
        assert self.napp.loop_manager.handle_topology_loaded.call_count == 1
        assert self.napp.load_liveness.call_count == 1

    def test_publish_liveness_status(self) -> None:
        """Test publish_liveness_status."""
        self.napp.controller.buffers.app.put = MagicMock()
        event_suffix, interfaces = "up", [MagicMock(id=1), MagicMock(id=2)]
        self.napp.publish_liveness_status(event_suffix, interfaces)
        assert self.napp.controller.buffers.app.put.call_count == 1
        event = self.napp.controller.buffers.app.put.call_args[0][0]
        assert event.name == f"kytos/of_lldp.liveness.{event_suffix}"
        assert event.content["interfaces"] == interfaces

    def test_get_interfaces(self):
        """Test _get_interfaces method."""
        expected_interfaces = self.get_topology_interfaces()

        interfaces = self.napp._get_interfaces()

        self.assertEqual(interfaces, expected_interfaces)

    def test_get_interfaces_dict(self):
        """Test _get_interfaces_dict method."""
        interfaces = self.napp._get_interfaces()
        expected_interfaces = {inter.id: inter for inter in interfaces}

        interfaces_dict = self.napp._get_interfaces_dict(interfaces)

        self.assertEqual(interfaces_dict, expected_interfaces)

    def test_get_lldp_interfaces(self):
        """Test _get_lldp_interfaces method."""
        lldp_interfaces = self.napp._get_lldp_interfaces()

        expected_interfaces = ['00:00:00:00:00:00:00:01:1',
                               '00:00:00:00:00:00:00:01:2',
                               '00:00:00:00:00:00:00:02:1',
                               '00:00:00:00:00:00:00:02:2']

        self.assertEqual(lldp_interfaces, expected_interfaces)

    def test_rest_get_lldp_interfaces(self):
        """Test get_lldp_interfaces method."""
        api = get_test_client(self.napp.controller, self.napp)
        url = f'{self.server_name_url}/v1/interfaces'
        response = api.open(url, method='GET')

        expected_data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
                                        '00:00:00:00:00:00:00:01:2',
                                        '00:00:00:00:00:00:00:02:1',
                                        '00:00:00:00:00:00:00:02:2']}
        self.assertEqual(response.json, expected_data)
        self.assertEqual(response.status_code, 200)

    def test_enable_disable_lldp_200(self):
        """Test 200 response for enable_lldp and disable_lldp methods."""
        data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
                               '00:00:00:00:00:00:00:01:2',
                               '00:00:00:00:00:00:00:02:1',
                               '00:00:00:00:00:00:00:02:2']}

        api = get_test_client(self.napp.controller, self.napp)
        self.napp.publish_liveness_status = MagicMock()

        url = f'{self.server_name_url}/v1/interfaces/disable'
        disable_response = api.open(url, method='POST', json=data)
        assert self.napp.liveness_controller.disable_interfaces.call_count == 1
        assert self.napp.publish_liveness_status.call_count == 1

        url = f'{self.server_name_url}/v1/interfaces/enable'
        enable_response = api.open(url, method='POST', json=data)

        self.assertEqual(disable_response.status_code, 200)
        self.assertEqual(enable_response.status_code, 200)

    def test_enable_disable_lldp_404(self):
        """Test 404 response for enable_lldp and disable_lldp methods."""
        data = {"interfaces": []}

        self.napp.controller.switches = {}
        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v1/interfaces/disable'
        disable_response = api.open(url, method='POST', json=data)

        url = f'{self.server_name_url}/v1/interfaces/enable'
        enable_response = api.open(url, method='POST', json=data)

        self.assertEqual(disable_response.status_code, 404)
        self.assertEqual(enable_response.status_code, 404)

    def test_enable_disable_lldp_400(self):
        """Test 400 response for enable_lldp and disable_lldp methods."""
        data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
                               '00:00:00:00:00:00:00:01:2',
                               '00:00:00:00:00:00:00:02:1',
                               '00:00:00:00:00:00:00:02:2',
                               '00:00:00:00:00:00:00:03:1',
                               '00:00:00:00:00:00:00:03:2',
                               '00:00:00:00:00:00:00:04:1']}

        api = get_test_client(self.napp.controller, self.napp)
        self.napp.publish_liveness_status = MagicMock()

        url = f'{self.server_name_url}/v1/interfaces/disable'
        disable_response = api.open(url, method='POST', json=data)
        assert self.napp.publish_liveness_status.call_count == 1

        url = f'{self.server_name_url}/v1/interfaces/enable'
        enable_response = api.open(url, method='POST', json=data)

        self.assertEqual(disable_response.status_code, 400)
        self.assertEqual(enable_response.status_code, 400)

    def test_get_time(self):
        """Test get polling time."""
        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v1/polling_time'
        response = api.open(url, method='GET')

        self.assertEqual(response.status_code, 200)

    def test_set_time(self):
        """Test update polling time."""
        data = {"polling_time": 5}

        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v1/polling_time'
        response = api.open(url, method='POST', json=data)

        self.assertEqual(response.status_code, 200)
        self.assertEqual(self.napp.polling_time, data['polling_time'])

    def test_set_time_400(self):
        """Test fail case the update polling time."""
        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v1/polling_time'

        data = {'polling_time': 'A'}
        response = api.open(url, method='POST', json=data)
        self.assertEqual(response.status_code, 400)

    def test_endpoint_enable_liveness(self):
        """Test POST v1/liveness/enable."""
        self.napp.liveness_manager.enable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        api = get_test_client(self.napp.controller, self.napp)
        url = f"{self.server_name_url}/v1/liveness/enable"
        data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
        response = api.open(url, method="POST", json=data)
        assert self.napp.liveness_controller.enable_interfaces.call_count == 1
        assert self.napp.liveness_manager.enable.call_count == 1
        assert self.napp.publish_liveness_status.call_count == 1
        assert response.json == {}
        assert response.status_code == 200

    def test_endpoint_disable_liveness(self):
        """Test POST v1/liveness/disable."""
        self.napp.liveness_manager.disable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        api = get_test_client(self.napp.controller, self.napp)
        url = f"{self.server_name_url}/v1/liveness/disable"
        data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
        response = api.open(url, method='POST', json=data)
        assert self.napp.liveness_controller.disable_interfaces.call_count == 1
        assert self.napp.liveness_manager.disable.call_count == 1
        assert self.napp.publish_liveness_status.call_count == 1
        assert response.json == {}
        assert response.status_code == 200

    def test_endpoint_get_liveness(self):
        """Test GET v1/liveness/."""
        self.napp.liveness_manager.enable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        api = get_test_client(self.napp.controller, self.napp)
        url = f"{self.server_name_url}/v1/liveness/"
        response = api.open(url, method="GET")
        assert response.json == {"interfaces": []}
        assert response.status_code == 200

    def test_endpoint_get_pair_liveness(self):
        """Test GET v1/liveness//pair."""
        self.napp.liveness_manager.enable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        api = get_test_client(self.napp.controller, self.napp)
        url = f"{self.server_name_url}/v1/liveness/pair"
        response = api.open(url, method="GET")
        assert response.json == {"pairs": []}
        assert response.status_code == 200


1			"""Test Main methods."""
2	1		from unittest import TestCase
3			from unittest.mock import AsyncMock, MagicMock, call, patch
4	1
5			from kytos.lib.helpers import (get_controller_mock, get_kytos_event_mock,
6			get_switch_mock, get_test_client)
7	1
8	1		from kytos.core.events import KytosEvent
9	1		from napps.kytos.of_lldp.utils import get_cookie
10			from tests.helpers import get_topology_mock
11
12	1
13	1	View Code Duplication	@patch('kytos.core.controller.Controller.get_switch_by_dpid')
			0 ignored issues – show Duplication introduced 2022-11-18 17:10 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
14	1		@patch('napps.kytos.of_lldp.main.Main._unpack_non_empty')
15	1		@patch('napps.kytos.of_lldp.main.UBInt32')
16	1		@patch('napps.kytos.of_lldp.main.DPID')
17	1		@patch('napps.kytos.of_lldp.main.LLDP')
18	1		@patch('napps.kytos.of_lldp.main.Ethernet')
19			async def test_on_ofpt_packet_in(*args):
20	1		"""Test on_ofpt_packet_in."""
21			(mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
22			mock_unpack_non_empty, mock_get_switch_by_dpid) = args
23
24	1		# pylint: disable=bad-option-value, import-outside-toplevel
25	1		from napps.kytos.of_lldp.main import Main
26	1		Main.get_liveness_controller = MagicMock()
27	1		topology = get_topology_mock()
28	1		controller = get_controller_mock()
29	1		controller.buffers.app.aput = AsyncMock()
30	1		controller.switches = topology.switches
31	1		napp = Main(controller)
32	1		napp.loop_manager.process_if_looped = AsyncMock()
33			napp.liveness_manager.consume_hello_if_enabled = AsyncMock()
34	1
35	1		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
36	1		message = MagicMock(in_port=1, data='data')
37			event = KytosEvent('ofpt_packet_in', content={'source': switch.connection,
38			'message': message})
39	1
40	1		mocked, ethernet, lldp, dpid, port_b = [MagicMock() for _ in range(5)]
41	1		mocked.value = 1
42	1		mock_ubint32.return_value = mocked
43	1		ethernet.ether_type = 0x88CC
44	1		ethernet.data = 'eth_data'
45	1		lldp.chassis_id.sub_value = 'chassis_id'
46	1		lldp.port_id.sub_value = 'port_id'
47	1		dpid.value = "00:00:00:00:00:00:00:02"
48			port_b.value = 2
49	1
50	1		mock_unpack_non_empty.side_effect = [ethernet, lldp, dpid, port_b]
51			mock_get_switch_by_dpid.return_value = get_switch_mock(dpid.value,
52	1		0x04)
53			await napp.on_ofpt_packet_in(event)
54	1
55			calls = [call(mock_ethernet, message.data),
56			call(mock_lldp, ethernet.data),
57			call(mock_dpid, lldp.chassis_id.sub_value),
58	1		call(mock_ubint32, lldp.port_id.sub_value)]
59	1		mock_unpack_non_empty.assert_has_calls(calls)
60	1		assert napp.loop_manager.process_if_looped.call_count == 1
61	1		assert napp.liveness_manager.consume_hello_if_enabled.call_count == 1
62			assert controller.buffers.app.aput.call_count == 1
63
64	1
65	1	View Code Duplication	@patch('kytos.core.controller.Controller.get_switch_by_dpid')
			0 ignored issues – show Duplication introduced 2022-11-18 17:10 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
66	1		@patch('napps.kytos.of_lldp.main.Main._unpack_non_empty')
67	1		@patch('napps.kytos.of_lldp.main.UBInt32')
68	1		@patch('napps.kytos.of_lldp.main.DPID')
69	1		@patch('napps.kytos.of_lldp.main.LLDP')
70	1		@patch('napps.kytos.of_lldp.main.Ethernet')
71			async def test_on_ofpt_packet_in_early_intf(*args):
72	1		"""Test on_ofpt_packet_in early intf return."""
73			(mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
74			mock_unpack_non_empty, mock_get_switch_by_dpid) = args
75
76	1		# pylint: disable=bad-option-value, import-outside-toplevel
77	1		from napps.kytos.of_lldp.main import Main
78	1		Main.get_liveness_controller = MagicMock()
79	1		topology = get_topology_mock()
80	1		controller = get_controller_mock()
81	1		controller.buffers.app.aput = AsyncMock()
82	1		controller.switches = topology.switches
83	1		napp = Main(controller)
84	1		napp.loop_manager.process_if_looped = AsyncMock()
85			napp.liveness_manager.consume_hello_if_enabled = AsyncMock()
86	1
87	1		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
88	1		message = MagicMock(in_port=1, data='data')
89			event = KytosEvent('ofpt_packet_in', content={'source': switch.connection,
90			'message': message})
91	1
92	1		mocked, ethernet, lldp, dpid, port_b = [MagicMock() for _ in range(5)]
93	1		mocked.value = 1
94	1		mock_ubint32.return_value = mocked
95	1		ethernet.ether_type = 0x88CC
96	1		ethernet.data = 'eth_data'
97	1		lldp.chassis_id.sub_value = 'chassis_id'
98	1		lldp.port_id.sub_value = 'port_id'
99	1		dpid.value = "00:00:00:00:00:00:00:02"
100			port_b.value = 2
101	1
102	1		mock_unpack_non_empty.side_effect = [ethernet, lldp, dpid, port_b]
103			mock_get_switch_by_dpid.return_value = get_switch_mock(dpid.value,
104	1		0x04)
105	1		switch.get_interface_by_port_no = MagicMock(return_value=None)
106			await napp.on_ofpt_packet_in(event)
107	1
108			calls = [call(mock_ethernet, message.data),
109			call(mock_lldp, ethernet.data),
110			call(mock_dpid, lldp.chassis_id.sub_value),
111	1		call(mock_ubint32, lldp.port_id.sub_value)]
112	1		mock_unpack_non_empty.assert_has_calls(calls)
113			switch.get_interface_by_port_no.assert_called()
114	1		# early return shouldn't allow these to get called
115	1		assert napp.loop_manager.process_if_looped.call_count == 0
116	1		assert napp.liveness_manager.consume_hello_if_enabled.call_count == 0
117			assert controller.buffers.app.aput.call_count == 0
118
119	1
120			# pylint: disable=protected-access,too-many-public-methods
121			class TestMain(TestCase):
122	1		"""Tests for the Main class."""
123	1
124	1		def setUp(self):
125	1		"""Execute steps before each tests."""
126			self.server_name_url = 'http://127.0.0.1:8181/api/kytos/of_lldp'
127
128	1		patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
129	1		# pylint: disable=bad-option-value, import-outside-toplevel
130			from napps.kytos.of_lldp.main import Main
131	1		self.addCleanup(patch.stopall)
132	1		Main.get_liveness_controller = MagicMock()
133	1		self.topology = get_topology_mock()
134			controller = get_controller_mock()
135			controller.switches = self.topology.switches
136	1
137	1		self.napp = Main(controller)
138
139	1		def get_topology_interfaces(self):
140	1		"""Return interfaces present in topology."""
141			interfaces = []
142			for switch in list(self.topology.switches.values()):
143			interfaces += list(switch.interfaces.values())
144	1		return interfaces
145
146			@patch('napps.kytos.of_lldp.main.of_msg_prio')
147	1		@patch('kytos.core.buffers.KytosEventBuffer.put')
148			@patch('napps.kytos.of_lldp.main.KytosEvent')
149			@patch('napps.kytos.of_lldp.main.VLAN')
150			@patch('napps.kytos.of_lldp.main.Ethernet')
151	1		@patch('napps.kytos.of_lldp.main.DPID')
152	1		@patch('napps.kytos.of_lldp.main.LLDP')
153	1		def test_execute(self, *args):
154	1		"""Test execute method."""
155	1		(_, _, mock_ethernet, _, mock_kytos_event,
156	1		mock_buffer_put, mock_of_msg_prio) = args
157	1
158	1		ethernet = MagicMock()
159			ethernet.pack.return_value = 'pack'
160	1		interfaces = self.get_topology_interfaces()
161			po_args = [(interface.switch.connection.protocol.version,
162	1		interface.port_number, 'pack') for interface in interfaces]
163
164	1		mock_ethernet.return_value = ethernet
165			mock_kytos_event.side_effect = po_args
166	1
167	1		mock_publish_stopped = MagicMock()
168	1		self.napp.try_to_publish_stopped_loops = mock_publish_stopped
169	1		self.napp.execute()
170
171	1		mock_of_msg_prio.assert_called()
172	1		mock_buffer_put.assert_has_calls([call(arg)
173	1		for arg in po_args])
174	1		mock_publish_stopped.assert_called()
175	1
176	1		@patch('requests.delete')
177	1		@patch('requests.post')
178			def test_handle_lldp_flows(self, mock_post, mock_delete):
179	1		"""Test handle_lldp_flow method."""
180	1		dpid = "00:00:00:00:00:00:00:01"
181	1		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
182			self.napp.controller.switches = {dpid: switch}
183	1		event_post = get_kytos_event_mock(name='kytos/topology.switch.enabled',
184	1		content={'dpid': dpid})
185	1
186	1		event_del = get_kytos_event_mock(name='kytos/topology.switch.disabled',
187			content={'dpid': dpid})
188
189	1		mock_post.return_value = MagicMock(status_code=202)
190	1		mock_delete.return_value = MagicMock(status_code=202)
191
192	1		self.napp._handle_lldp_flows(event_post)
193	1		mock_post.assert_called()
194	1
195			self.napp._handle_lldp_flows(event_del)
196	1		mock_delete.assert_called()
197	1
198			@patch("time.sleep")
199	1		@patch("requests.post")
200			def test_handle_lldp_flows_retries(self, mock_post, _):
201	1		"""Test handle_lldp_flow method retries."""
202	1		dpid = "00:00:00:00:00:00:00:01"
203	1		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
204			self.napp.controller.switches = {dpid: switch}
205	1		event_post = get_kytos_event_mock(name="kytos/topology.switch.enabled",
206	1		content={"dpid": dpid})
207	1
208	1		mock = MagicMock()
209			mock.request.method = "POST"
210			mock.status_code = 500
211	1		mock.text = "some_err"
212			mock_post.return_value = mock
213			self.napp._handle_lldp_flows(event_post)
214	1		self.assertTrue(mock_post.call_count, 3)
215	1
216			@patch('napps.kytos.of_lldp.main.PO13')
217	1		@patch('napps.kytos.of_lldp.main.AO13')
218	1		def test_build_lldp_packet_out(self, *args):
219			"""Test _build_lldp_packet_out method."""
220	1		(mock_ao13, mock_po13) = args
221	1
222			ao13 = MagicMock()
223	1		po13 = MagicMock()
224	1		po13.actions = []
225	1
226			mock_ao13.return_value = ao13
227	1		mock_po13.return_value = po13
228	1
229	1		packet_out13 = self.napp._build_lldp_packet_out(0x04, 2, 'data2')
230	1		packet_out14 = self.napp._build_lldp_packet_out(0x05, 3, 'data3')
231
232			self.assertEqual(packet_out13.data, 'data2')
233	1		self.assertEqual(packet_out13.actions, [ao13])
234	1		self.assertEqual(packet_out13.actions[0].port, 2)
235	1
236	1		self.assertIsNone(packet_out14)
237	1
238	1		@patch('napps.kytos.of_lldp.main.settings')
239	1		@patch('napps.kytos.of_lldp.main.EtherType')
240			@patch('napps.kytos.of_lldp.main.Port13')
241	1		def test_build_lldp_flow(self, *args):
242	1		"""Test _build_lldp_flow method."""
243	1		(mock_v0x04_port, mock_ethertype,
244			mock_settings) = args
245	1		self.napp.vlan_id = None
246			mock_v0x04_port.OFPP_CONTROLLER = 1234
247	1
248	1		mock_ethertype.LLDP = 10
249	1		mock_settings.FLOW_VLAN_VID = None
250			mock_settings.FLOW_PRIORITY = 1500
251	1		mock_settings.TABLE_ID = 0
252	1		dpid = "00:00:00:00:00:00:00:01"
253
254	1		flow = {}
255	1		match = {}
256			flow['priority'] = 1500
257	1		flow['table_id'] = 0
258	1		match['dl_type'] = 10
259	1
260	1		flow['match'] = match
261			expected_flow_v0x04 = flow.copy()
262	1		expected_flow_v0x04['cookie'] = get_cookie(dpid)
263	1		expected_flow_v0x04['cookie_mask'] = 0xffffffffffffffff
264	1
265	1		expected_flow_v0x04['actions'] = [{'action_type': 'output',
266			'port': 1234}]
267	1
268			flow_mod10 = self.napp._build_lldp_flow(0x01, get_cookie(dpid))
269	1		flow_mod13 = self.napp._build_lldp_flow(0x04, get_cookie(dpid))
270	1
271			self.assertIsNone(flow_mod10)
272	1		self.assertDictEqual(flow_mod13, expected_flow_v0x04)
273	1
274	1		def test_unpack_non_empty(self):
275	1		"""Test _unpack_non_empty method."""
276	1		desired_class = MagicMock()
277			data = MagicMock()
278	1		data.value = 'data'
279	1
280	1		obj = self.napp._unpack_non_empty(desired_class, data)
281	1
282	1		obj.unpack.assert_called_with('data')
283
284	1		def test_get_data(self):
285	1		"""Test _get_data method."""
286	1		req = MagicMock()
287	1		interfaces = ['00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:01:2']
288			req.get_json.return_value = {'interfaces': interfaces}
289	1
290			data = self.napp._get_data(req)
291	1
292	1		self.assertEqual(data, interfaces)
293
294	1		def test_load_liveness(self) -> None:
295	1		"""Test load_liveness."""
296			self.napp.load_liveness()
297	1		count = self.napp.liveness_controller.get_enabled_interfaces.call_count
298	1		assert count == 1
299
300	1		def test_handle_topology_loaded(self) -> None:
301			"""Test handle_topology_loaded."""
302	1		event = KytosEvent("kytos/topology.topology_loaded",
303	1		content={"topology": {}})
304	1		self.napp.load_liveness = MagicMock()
305			self.napp.loop_manager.handle_topology_loaded = MagicMock()
306	1		self.napp.handle_topology_loaded(event)
307			assert self.napp.loop_manager.handle_topology_loaded.call_count == 1
308	1		assert self.napp.load_liveness.call_count == 1
309
310	1		def test_publish_liveness_status(self) -> None:
311			"""Test publish_liveness_status."""
312	1		self.napp.controller.buffers.app.put = MagicMock()
313	1		event_suffix, interfaces = "up", [MagicMock(id=1), MagicMock(id=2)]
314			self.napp.publish_liveness_status(event_suffix, interfaces)
315	1		assert self.napp.controller.buffers.app.put.call_count == 1
316	1		event = self.napp.controller.buffers.app.put.call_args[0][0]
317			assert event.name == f"kytos/of_lldp.liveness.{event_suffix}"
318	1		assert event.content["interfaces"] == interfaces
319
320	1		def test_get_interfaces(self):
321	1		"""Test _get_interfaces method."""
322	1		expected_interfaces = self.get_topology_interfaces()
323
324	1		interfaces = self.napp._get_interfaces()
325
326	1		self.assertEqual(interfaces, expected_interfaces)
327
328	1		def test_get_interfaces_dict(self):
329	1		"""Test _get_interfaces_dict method."""
330	1		interfaces = self.napp._get_interfaces()
331	1		expected_interfaces = {inter.id: inter for inter in interfaces}
332	1
333			interfaces_dict = self.napp._get_interfaces_dict(interfaces)
334	1
335			self.assertEqual(interfaces_dict, expected_interfaces)
336	1
337	1		def test_get_lldp_interfaces(self):
338	1		"""Test _get_lldp_interfaces method."""
339	1		lldp_interfaces = self.napp._get_lldp_interfaces()
340	1
341	1		expected_interfaces = ['00:00:00:00:00:00:00:01:1',
342	1		'00:00:00:00:00:00:00:01:2',
343			'00:00:00:00:00:00:00:02:1',
344	1		'00:00:00:00:00:00:00:02:2']
345
346	1		self.assertEqual(lldp_interfaces, expected_interfaces)
347	1
348	1		def test_rest_get_lldp_interfaces(self):
349			"""Test get_lldp_interfaces method."""
350	1		api = get_test_client(self.napp.controller, self.napp)
351			url = f'{self.server_name_url}/v1/interfaces'
352	1		response = api.open(url, method='GET')
353	1
354	1		expected_data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
355	1		'00:00:00:00:00:00:00:01:2',
356			'00:00:00:00:00:00:00:02:1',
357	1		'00:00:00:00:00:00:00:02:2']}
358			self.assertEqual(response.json, expected_data)
359	1		self.assertEqual(response.status_code, 200)
360	1
361			def test_enable_disable_lldp_200(self):
362			"""Test 200 response for enable_lldp and disable_lldp methods."""
363			data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
364	1		'00:00:00:00:00:00:00:01:2',
365			'00:00:00:00:00:00:00:02:1',
366	1		'00:00:00:00:00:00:00:02:2']}
367
368	1		api = get_test_client(self.napp.controller, self.napp)
369	1		self.napp.publish_liveness_status = MagicMock()
370	1
371			url = f'{self.server_name_url}/v1/interfaces/disable'
372			disable_response = api.open(url, method='POST', json=data)
373			assert self.napp.liveness_controller.disable_interfaces.call_count == 1
374	1		assert self.napp.publish_liveness_status.call_count == 1
375	1
376			url = f'{self.server_name_url}/v1/interfaces/enable'
377	1		enable_response = api.open(url, method='POST', json=data)
378
379	1		self.assertEqual(disable_response.status_code, 200)
380			self.assertEqual(enable_response.status_code, 200)
381
382			def test_enable_disable_lldp_404(self):
383	1		"""Test 404 response for enable_lldp and disable_lldp methods."""
384	1		data = {"interfaces": []}
385	1
386	1		self.napp.controller.switches = {}
387	1		api = get_test_client(self.napp.controller, self.napp)
388	1
389	1		url = f'{self.server_name_url}/v1/interfaces/disable'
390	1		disable_response = api.open(url, method='POST', json=data)
391	1
392	1		url = f'{self.server_name_url}/v1/interfaces/enable'
393			enable_response = api.open(url, method='POST', json=data)
394	1
395			self.assertEqual(disable_response.status_code, 404)
396	1		self.assertEqual(enable_response.status_code, 404)
397	1
398	1		def test_enable_disable_lldp_400(self):
399	1		"""Test 400 response for enable_lldp and disable_lldp methods."""
400	1		data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
401	1		'00:00:00:00:00:00:00:01:2',
402	1		'00:00:00:00:00:00:00:02:1',
403	1		'00:00:00:00:00:00:00:02:2',
404			'00:00:00:00:00:00:00:03:1',
405	1		'00:00:00:00:00:00:00:03:2',
406			'00:00:00:00:00:00:00:04:1']}
407	1
408			api = get_test_client(self.napp.controller, self.napp)
409			self.napp.publish_liveness_status = MagicMock()
410
411			url = f'{self.server_name_url}/v1/interfaces/disable'
412			disable_response = api.open(url, method='POST', json=data)
413			assert self.napp.publish_liveness_status.call_count == 1
414	1
415	1		url = f'{self.server_name_url}/v1/interfaces/enable'
416	1		enable_response = api.open(url, method='POST', json=data)
417	1
418	1		self.assertEqual(disable_response.status_code, 400)
419	1		self.assertEqual(enable_response.status_code, 400)
420
421	1		def test_get_time(self):
422	1		"""Test get polling time."""
423	1		api = get_test_client(self.napp.controller, self.napp)
424
425	1		url = f'{self.server_name_url}/v1/polling_time'
426			response = api.open(url, method='GET')
427	1
428	1		self.assertEqual(response.status_code, 200)
429	1
430			def test_set_time(self):
431	1		"""Test update polling time."""
432			data = {"polling_time": 5}
433	1
434	1		api = get_test_client(self.napp.controller, self.napp)
435	1
436	1		url = f'{self.server_name_url}/v1/polling_time'
437			response = api.open(url, method='POST', json=data)
438	1
439			self.assertEqual(response.status_code, 200)
440	1		self.assertEqual(self.napp.polling_time, data['polling_time'])
441	1
442	1		def test_set_time_400(self):
443	1		"""Test fail case the update polling time."""
444			api = get_test_client(self.napp.controller, self.napp)
445	1
446			url = f'{self.server_name_url}/v1/polling_time'
447	1
448	1		data = {'polling_time': 'A'}
449	1		response = api.open(url, method='POST', json=data)
450	1		self.assertEqual(response.status_code, 400)
451	1
452	1		def test_endpoint_enable_liveness(self):
453	1		"""Test POST v1/liveness/enable."""
454	1		self.napp.liveness_manager.enable = MagicMock()
455	1		self.napp.publish_liveness_status = MagicMock()
456	1		api = get_test_client(self.napp.controller, self.napp)
457	1		url = f"{self.server_name_url}/v1/liveness/enable"
458			data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
459	1		response = api.open(url, method="POST", json=data)
460			assert self.napp.liveness_controller.enable_interfaces.call_count == 1
461	1		assert self.napp.liveness_manager.enable.call_count == 1
462	1		assert self.napp.publish_liveness_status.call_count == 1
463	1		assert response.json == {}
464	1		assert response.status_code == 200
465	1
466	1		def test_endpoint_disable_liveness(self):
467	1		"""Test POST v1/liveness/disable."""
468	1		self.napp.liveness_manager.disable = MagicMock()
469	1		self.napp.publish_liveness_status = MagicMock()
470	1		api = get_test_client(self.napp.controller, self.napp)
471	1		url = f"{self.server_name_url}/v1/liveness/disable"
472			data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
473	1		response = api.open(url, method='POST', json=data)
474			assert self.napp.liveness_controller.disable_interfaces.call_count == 1
475	1		assert self.napp.liveness_manager.disable.call_count == 1
476	1		assert self.napp.publish_liveness_status.call_count == 1
477	1		assert response.json == {}
478	1		assert response.status_code == 200
479	1
480	1		def test_endpoint_get_liveness(self):
481			"""Test GET v1/liveness/."""
482	1		self.napp.liveness_manager.enable = MagicMock()
483			self.napp.publish_liveness_status = MagicMock()
484	1		api = get_test_client(self.napp.controller, self.napp)
485	1		url = f"{self.server_name_url}/v1/liveness/"
486	1		response = api.open(url, method="GET")
487	1		assert response.json == {"interfaces": []}
488	1		assert response.status_code == 200
489	1
490			def test_endpoint_get_pair_liveness(self):
491	1		"""Test GET v1/liveness//pair."""
492			self.napp.liveness_manager.enable = MagicMock()
493	1		self.napp.publish_liveness_status = MagicMock()
494	1		api = get_test_client(self.napp.controller, self.napp)
495	1		url = f"{self.server_name_url}/v1/liveness/pair"
496	1		response = api.open(url, method="GET")
497	1		assert response.json == {"pairs": []}
498			assert response.status_code == 200
499

kytos-ng / of_lldp

Pull Request — master (#88)

TestMain.test_set_time() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like