build.tests.unit.test_main.TestMain.teardown_method() - Code Metrics - Inspection of "Merge pull request #78 from kytos-ng/chore/starlet..." - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 33e026...27d34c )

by Vinicius

created 2023-05-03 17:58 UTC

TestMain.teardown_method() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

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

"""Test Main methods."""
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


async def test_on_table_enabled():
    """Test on_table_enabled"""
    # pylint: disable=bad-option-value, import-outside-toplevel
    from napps.kytos.of_lldp.main import Main
    controller = get_controller_mock()
    controller.buffers.app.aput = AsyncMock()
    napp = Main(controller)

    # Succesfully setting table groups
    content = {"of_lldp": {"base": 123}}
    event = KytosEvent(name="kytos/of_multi_table.enable_table",
                       content=content)
    await napp.on_table_enabled(event)
    assert napp.table_group == content["of_lldp"]
    assert controller.buffers.app.aput.call_count == 1

    # Failure at setting table groups
    content = {"of_lldp": {"unknown": 123}}
    event = KytosEvent(name="kytos/of_multi_table.enable_table",
                       content=content)
    await napp.on_table_enabled(event)
    assert controller.buffers.app.aput.call_count == 1


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

    def setup_method(self):
        """Execute steps before each tests."""
        # 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
        Main.get_liveness_controller = MagicMock()
        self.topology = get_topology_mock()
        controller = get_controller_mock()
        controller.switches = self.topology.switches
        self.base_endpoint = "kytos/of_lldp/v1"
        self.napp = Main(controller)
        self.api_client = get_test_client(controller, self.napp)

    def teardown_method(self) -> None:
        """Teardown."""
        patch.stopall()

    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('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_of_msg_prio) = args
        mock_buffer_put = MagicMock()
        self.napp.controller.buffers.msg_out.put = mock_buffer_put

        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)
        assert 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')

        assert packet_out13.data == 'data2'
        assert packet_out13.actions == [ao13]
        assert packet_out13.actions[0].port == 2
        assert packet_out14 is None

    @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}]
        expected_flow_v0x04['table_group'] = 'base'
        expected_flow_v0x04['owner'] = 'of_lldp'

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

        assert flow_mod10 is None
        assert 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, monkeypatch):
        """Test _get_data method."""
        interfaces = ['00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:01:2']
        monkeypatch.setattr("napps.kytos.of_lldp.main.get_json_or_400",
                            lambda req, loop: {"interfaces": interfaces})
        data = self.napp._get_data(MagicMock())
        assert 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()
        assert 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)
        assert 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']
        assert lldp_interfaces == expected_interfaces

    async def test_rest_get_lldp_interfaces(self):
        """Test get_lldp_interfaces method."""
        endpoint = f"{self.base_endpoint}/interfaces"
        response = await self.api_client.get(endpoint)
        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']}
        assert response.status_code == 200
        assert response.json() == expected_data

    async def test_enable_disable_lldp_200(self, event_loop):
        """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']}
        self.napp.controller.loop = event_loop
        self.napp.publish_liveness_status = MagicMock()
        endpoint = f"{self.base_endpoint}/interfaces/disable"
        response = await self.api_client.post(endpoint, json=data)
        assert response.status_code == 200
        assert self.napp.liveness_controller.disable_interfaces.call_count == 1
        assert self.napp.publish_liveness_status.call_count == 1
        endpoint = f"{self.base_endpoint}/interfaces/enable"
        response = await self.api_client.post(endpoint, json=data)
        assert response.status_code == 200

    async def test_enable_disable_lldp_404(self):
        """Test 404 response for enable_lldp and disable_lldp methods."""
        data = {"interfaces": []}
        self.napp.controller.switches = {}
        endpoint = f"{self.base_endpoint}/disable"
        response = await self.api_client.post(endpoint, json=data)
        assert response.status_code == 404
        endpoint = f"{self.base_endpoint}/enable"
        response = await self.api_client.post(endpoint, json=data)
        assert response.status_code == 404

    async def test_enable_disable_lldp_400(self, event_loop):
        """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']}
        self.napp.controller.loop = event_loop
        self.napp.publish_liveness_status = MagicMock()
        url = f'{self.base_endpoint}/interfaces/disable'
        response = await self.api_client.post(url, json=data)
        assert response.status_code == 400
        assert self.napp.publish_liveness_status.call_count == 1

        url = f'{self.base_endpoint}/interfaces/enable'
        response = await self.api_client.post(url, json=data)
        assert response.status_code == 400

    async def test_get_time(self):
        """Test get polling time."""
        url = f"{self.base_endpoint}/polling_time"
        response = await self.api_client.get(url)
        assert response.status_code == 200

    async def test_set_polling_time(self):
        """Test update polling time."""
        url = f"{self.base_endpoint}/polling_time"
        data = {'polling_time': 5}
        response = await self.api_client.post(url, json=data)
        assert response.status_code == 200

    async def test_set_time_400(self):
        """Test fail case the update polling time."""
        url = f"{self.base_endpoint}/polling_time"
        data = {'polling_time': 'A'}
        response = await self.api_client.post(url, json=data)
        assert response.status_code == 400

    async def test_endpoint_enable_liveness(self, event_loop):
        """Test POST v1/liveness/enable."""
        self.napp.controller.loop = event_loop
        self.napp.liveness_manager.enable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        url = f"{self.base_endpoint}/liveness/enable"
        data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
        response = await self.api_client.post(url, json=data)
        assert response.status_code == 200
        assert response.json() == {}
        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

    async def test_endpoint_disable_liveness(self, event_loop):
        """Test POST v1/liveness/disable."""
        self.napp.controller.loop = event_loop
        self.napp.liveness_manager.disable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        url = f"{self.base_endpoint}/liveness/disable"
        data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
        response = await self.api_client.post(url, json=data)
        assert response.status_code == 200
        assert response.json() == {}
        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

    async def test_endpoint_get_liveness(self):
        """Test GET v1/liveness/."""
        self.napp.liveness_manager.enable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        url = f"{self.base_endpoint}/liveness/"
        response = await self.api_client.get(url)
        assert response.status_code == 200
        assert response.json() == {"interfaces": []}

    async def test_endpoint_get_pair_liveness(self):
        """Test GET v1/liveness//pair."""
        self.napp.liveness_manager.enable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        url = f"{self.base_endpoint}/liveness/pair"
        response = await self.api_client.get(url)
        assert response.status_code == 200
        assert response.json() == {"pairs": []}

    def test_set_flow_table_group_owner(self):
        """Test set_flow_table_group_owner"""
        self.napp.table_group = {"base": 2}
        flow = {}
        self.napp.set_flow_table_group_owner(flow, "base")
        assert "table_group" in flow
        assert "owner" in flow
        assert flow["table_id"] == 2


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

kytos-ng / of_lldp

Push — master ( 33e026...27d34c )

TestMain.teardown_method() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like