build.tests.unit.test_main.TestMain.teardown_method() - Code Metrics - Inspection of "chore: `@rest` endpoints are now run by `starlette..." - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#78)

by Vinicius

created 2023-04-21 18:27 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


# pylint: disable=protected-access,too-many-public-methods,
# pylint: disable=attribute-defined-outside-init
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}]

        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: {"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):
        """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.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):
        """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.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):
        """Test POST v1/liveness/enable."""
        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):
        """Test POST v1/liveness/disable."""
        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": []}


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			# pylint: disable=protected-access,too-many-public-methods,
120			# pylint: disable=attribute-defined-outside-init
121	1		class TestMain:
122			"""Tests for the Main class."""
123
124	1		def setup_method(self):
125			"""Execute steps before each tests."""
126			# patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
127			# pylint: disable=bad-option-value, import-outside-toplevel
128	1		from napps.kytos.of_lldp.main import Main
129	1		Main.get_liveness_controller = MagicMock()
130	1		self.topology = get_topology_mock()
131	1		controller = get_controller_mock()
132	1		controller.switches = self.topology.switches
133	1		self.base_endpoint = "kytos/of_lldp/v1"
134	1		self.napp = Main(controller)
135	1		self.api_client = get_test_client(controller, self.napp)
136
137	1		def teardown_method(self) -> None:
138			"""Teardown."""
139	1		patch.stopall()
140
141	1		def get_topology_interfaces(self):
142			"""Return interfaces present in topology."""
143	1		interfaces = []
144	1		for switch in list(self.topology.switches.values()):
145	1		interfaces += list(switch.interfaces.values())
146	1		return interfaces
147
148	1		@patch('napps.kytos.of_lldp.main.of_msg_prio')
149	1		@patch('napps.kytos.of_lldp.main.KytosEvent')
150	1		@patch('napps.kytos.of_lldp.main.VLAN')
151	1		@patch('napps.kytos.of_lldp.main.Ethernet')
152	1		@patch('napps.kytos.of_lldp.main.DPID')
153	1		@patch('napps.kytos.of_lldp.main.LLDP')
154	1		def test_execute(self, *args):
155			"""Test execute method."""
156	1		(_, _, mock_ethernet, _, mock_kytos_event, mock_of_msg_prio) = args
157	1		mock_buffer_put = MagicMock()
158	1		self.napp.controller.buffers.msg_out.put = mock_buffer_put
159
160	1		ethernet = MagicMock()
161	1		ethernet.pack.return_value = 'pack'
162	1		interfaces = self.get_topology_interfaces()
163	1		po_args = [(interface.switch.connection.protocol.version,
164			interface.port_number, 'pack') for interface in interfaces]
165
166	1		mock_ethernet.return_value = ethernet
167	1		mock_kytos_event.side_effect = po_args
168
169	1		mock_publish_stopped = MagicMock()
170	1		self.napp.try_to_publish_stopped_loops = mock_publish_stopped
171	1		self.napp.execute()
172
173	1		mock_of_msg_prio.assert_called()
174	1		mock_buffer_put.assert_has_calls([call(arg)
175			for arg in po_args])
176	1		mock_publish_stopped.assert_called()
177
178	1		@patch('requests.delete')
179	1		@patch('requests.post')
180	1		def test_handle_lldp_flows(self, mock_post, mock_delete):
181			"""Test handle_lldp_flow method."""
182	1		dpid = "00:00:00:00:00:00:00:01"
183	1		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
184	1		self.napp.controller.switches = {dpid: switch}
185	1		event_post = get_kytos_event_mock(name='kytos/topology.switch.enabled',
186			content={'dpid': dpid})
187
188	1		event_del = get_kytos_event_mock(name='kytos/topology.switch.disabled',
189			content={'dpid': dpid})
190
191	1		mock_post.return_value = MagicMock(status_code=202)
192	1		mock_delete.return_value = MagicMock(status_code=202)
193
194	1		self.napp._handle_lldp_flows(event_post)
195	1		mock_post.assert_called()
196
197	1		self.napp._handle_lldp_flows(event_del)
198	1		mock_delete.assert_called()
199
200	1		@patch("time.sleep")
201	1		@patch("requests.post")
202	1		def test_handle_lldp_flows_retries(self, mock_post, _):
203			"""Test handle_lldp_flow method retries."""
204	1		dpid = "00:00:00:00:00:00:00:01"
205	1		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
206	1		self.napp.controller.switches = {dpid: switch}
207	1		event_post = get_kytos_event_mock(name="kytos/topology.switch.enabled",
208			content={"dpid": dpid})
209
210	1		mock = MagicMock()
211	1		mock.request.method = "POST"
212	1		mock.status_code = 500
213	1		mock.text = "some_err"
214	1		mock_post.return_value = mock
215	1		self.napp._handle_lldp_flows(event_post)
216	1		assert mock_post.call_count == 3
217
218	1		@patch('napps.kytos.of_lldp.main.PO13')
219	1		@patch('napps.kytos.of_lldp.main.AO13')
220	1		def test_build_lldp_packet_out(self, *args):
221			"""Test _build_lldp_packet_out method."""
222	1		(mock_ao13, mock_po13) = args
223
224	1		ao13 = MagicMock()
225	1		po13 = MagicMock()
226	1		po13.actions = []
227
228	1		mock_ao13.return_value = ao13
229	1		mock_po13.return_value = po13
230
231	1		packet_out13 = self.napp._build_lldp_packet_out(0x04, 2, 'data2')
232	1		packet_out14 = self.napp._build_lldp_packet_out(0x05, 3, 'data3')
233
234	1		assert packet_out13.data == 'data2'
235	1		assert packet_out13.actions == [ao13]
236	1		assert packet_out13.actions[0].port == 2
237	1		assert packet_out14 is None
238
239	1		@patch('napps.kytos.of_lldp.main.settings')
240	1		@patch('napps.kytos.of_lldp.main.EtherType')
241	1		@patch('napps.kytos.of_lldp.main.Port13')
242	1		def test_build_lldp_flow(self, *args):
243			"""Test _build_lldp_flow method."""
244	1		(mock_v0x04_port, mock_ethertype,
245			mock_settings) = args
246	1		self.napp.vlan_id = None
247	1		mock_v0x04_port.OFPP_CONTROLLER = 1234
248
249	1		mock_ethertype.LLDP = 10
250	1		mock_settings.FLOW_VLAN_VID = None
251	1		mock_settings.FLOW_PRIORITY = 1500
252	1		mock_settings.TABLE_ID = 0
253	1		dpid = "00:00:00:00:00:00:00:01"
254
255	1		flow = {}
256	1		match = {}
257	1		flow['priority'] = 1500
258	1		flow['table_id'] = 0
259	1		match['dl_type'] = 10
260
261	1		flow['match'] = match
262	1		expected_flow_v0x04 = flow.copy()
263	1		expected_flow_v0x04['cookie'] = get_cookie(dpid)
264	1		expected_flow_v0x04['cookie_mask'] = 0xffffffffffffffff
265
266	1		expected_flow_v0x04['actions'] = [{'action_type': 'output',
267			'port': 1234}]
268
269	1		flow_mod10 = self.napp._build_lldp_flow(0x01, get_cookie(dpid))
270	1		flow_mod13 = self.napp._build_lldp_flow(0x04, get_cookie(dpid))
271
272	1		assert flow_mod10 is None
273	1		assert flow_mod13 == expected_flow_v0x04
274
275	1		def test_unpack_non_empty(self):
276			"""Test _unpack_non_empty method."""
277	1		desired_class = MagicMock()
278	1		data = MagicMock()
279	1		data.value = 'data'
280
281	1		obj = self.napp._unpack_non_empty(desired_class, data)
282
283	1		obj.unpack.assert_called_with('data')
284
285	1		def test_get_data(self, monkeypatch):
286			"""Test _get_data method."""
287	1		interfaces = ['00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:01:2']
288	1		monkeypatch.setattr("napps.kytos.of_lldp.main.get_json_or_400",
289			lambda req: {"interfaces": interfaces})
290	1		data = self.napp._get_data(MagicMock())
291	1		assert data == interfaces
292
293	1		def test_load_liveness(self) -> None:
294			"""Test load_liveness."""
295	1		self.napp.load_liveness()
296	1		count = self.napp.liveness_controller.get_enabled_interfaces.call_count
297	1		assert count == 1
298
299	1		def test_handle_topology_loaded(self) -> None:
300			"""Test handle_topology_loaded."""
301	1		event = KytosEvent("kytos/topology.topology_loaded",
302			content={"topology": {}})
303	1		self.napp.load_liveness = MagicMock()
304	1		self.napp.loop_manager.handle_topology_loaded = MagicMock()
305	1		self.napp.handle_topology_loaded(event)
306	1		assert self.napp.loop_manager.handle_topology_loaded.call_count == 1
307	1		assert self.napp.load_liveness.call_count == 1
308
309	1		def test_publish_liveness_status(self) -> None:
310			"""Test publish_liveness_status."""
311	1		self.napp.controller.buffers.app.put = MagicMock()
312	1		event_suffix, interfaces = "up", [MagicMock(id=1), MagicMock(id=2)]
313	1		self.napp.publish_liveness_status(event_suffix, interfaces)
314	1		assert self.napp.controller.buffers.app.put.call_count == 1
315	1		event = self.napp.controller.buffers.app.put.call_args[0][0]
316	1		assert event.name == f"kytos/of_lldp.liveness.{event_suffix}"
317	1		assert event.content["interfaces"] == interfaces
318
319	1		def test_get_interfaces(self):
320			"""Test _get_interfaces method."""
321	1		expected_interfaces = self.get_topology_interfaces()
322	1		interfaces = self.napp._get_interfaces()
323	1		assert interfaces == expected_interfaces
324
325	1		def test_get_interfaces_dict(self):
326			"""Test _get_interfaces_dict method."""
327	1		interfaces = self.napp._get_interfaces()
328	1		expected_interfaces = {inter.id: inter for inter in interfaces}
329	1		interfaces_dict = self.napp._get_interfaces_dict(interfaces)
330	1		assert interfaces_dict == expected_interfaces
331
332	1		def test_get_lldp_interfaces(self):
333			"""Test _get_lldp_interfaces method."""
334	1		lldp_interfaces = self.napp._get_lldp_interfaces()
335	1		expected_interfaces = ['00:00:00:00:00:00:00:01:1',
336			'00:00:00:00:00:00:00:01:2',
337			'00:00:00:00:00:00:00:02:1',
338			'00:00:00:00:00:00:00:02:2']
339	1		assert lldp_interfaces == expected_interfaces
340
341	1		async def test_rest_get_lldp_interfaces(self):
342			"""Test get_lldp_interfaces method."""
343	1		endpoint = f"{self.base_endpoint}/interfaces"
344	1		response = await self.api_client.get(endpoint)
345	1		expected_data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
346			'00:00:00:00:00:00:00:01:2',
347			'00:00:00:00:00:00:00:02:1',
348			'00:00:00:00:00:00:00:02:2']}
349	1		assert response.status_code == 200
350	1		assert response.json() == expected_data
351
352	1		async def test_enable_disable_lldp_200(self):
353			"""Test 200 response for enable_lldp and disable_lldp methods."""
354	1		data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
355			'00:00:00:00:00:00:00:01:2',
356			'00:00:00:00:00:00:00:02:1',
357			'00:00:00:00:00:00:00:02:2']}
358	1		self.napp.publish_liveness_status = MagicMock()
359	1		endpoint = f"{self.base_endpoint}/interfaces/disable"
360	1		response = await self.api_client.post(endpoint, json=data)
361	1		assert response.status_code == 200
362	1		assert self.napp.liveness_controller.disable_interfaces.call_count == 1
363	1		assert self.napp.publish_liveness_status.call_count == 1
364	1		endpoint = f"{self.base_endpoint}/interfaces/enable"
365	1		response = await self.api_client.post(endpoint, json=data)
366	1		assert response.status_code == 200
367
368	1		async def test_enable_disable_lldp_404(self):
369			"""Test 404 response for enable_lldp and disable_lldp methods."""
370	1		data = {"interfaces": []}
371	1		self.napp.controller.switches = {}
372	1		endpoint = f"{self.base_endpoint}/disable"
373	1		response = await self.api_client.post(endpoint, json=data)
374	1		assert response.status_code == 404
375	1		endpoint = f"{self.base_endpoint}/enable"
376	1		response = await self.api_client.post(endpoint, json=data)
377	1		assert response.status_code == 404
378
379	1		async def test_enable_disable_lldp_400(self):
380			"""Test 400 response for enable_lldp and disable_lldp methods."""
381	1		data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
382			'00:00:00:00:00:00:00:01:2',
383			'00:00:00:00:00:00:00:02:1',
384			'00:00:00:00:00:00:00:02:2',
385			'00:00:00:00:00:00:00:03:1',
386			'00:00:00:00:00:00:00:03:2',
387			'00:00:00:00:00:00:00:04:1']}
388	1		self.napp.publish_liveness_status = MagicMock()
389	1		url = f'{self.base_endpoint}/interfaces/disable'
390	1		response = await self.api_client.post(url, json=data)
391	1		assert response.status_code == 400
392	1		assert self.napp.publish_liveness_status.call_count == 1
393
394	1		url = f'{self.base_endpoint}/interfaces/enable'
395	1		response = await self.api_client.post(url, json=data)
396	1		assert response.status_code == 400
397
398	1		async def test_get_time(self):
399			"""Test get polling time."""
400	1		url = f"{self.base_endpoint}/polling_time"
401	1		response = await self.api_client.get(url)
402	1		assert response.status_code == 200
403
404	1		async def test_set_polling_time(self):
405			"""Test update polling time."""
406	1		url = f"{self.base_endpoint}/polling_time"
407	1		data = {'polling_time': 5}
408	1		response = await self.api_client.post(url, json=data)
409	1		assert response.status_code == 200
410
411	1		async def test_set_time_400(self):
412			"""Test fail case the update polling time."""
413	1		url = f"{self.base_endpoint}/polling_time"
414	1		data = {'polling_time': 'A'}
415	1		response = await self.api_client.post(url, json=data)
416	1		assert response.status_code == 400
417
418	1		async def test_endpoint_enable_liveness(self):
419			"""Test POST v1/liveness/enable."""
420	1		self.napp.liveness_manager.enable = MagicMock()
421	1		self.napp.publish_liveness_status = MagicMock()
422	1		url = f"{self.base_endpoint}/liveness/enable"
423	1		data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
424	1		response = await self.api_client.post(url, json=data)
425	1		assert response.status_code == 200
426	1		assert response.json() == {}
427	1		assert self.napp.liveness_controller.enable_interfaces.call_count == 1
428	1		assert self.napp.liveness_manager.enable.call_count == 1
429	1		assert self.napp.publish_liveness_status.call_count == 1
430
431	1		async def test_endpoint_disable_liveness(self):
432			"""Test POST v1/liveness/disable."""
433	1		self.napp.liveness_manager.disable = MagicMock()
434	1		self.napp.publish_liveness_status = MagicMock()
435	1		url = f"{self.base_endpoint}/liveness/disable"
436	1		data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
437	1		response = await self.api_client.post(url, json=data)
438	1		assert response.status_code == 200
439	1		assert response.json() == {}
440	1		assert self.napp.liveness_controller.disable_interfaces.call_count == 1
441	1		assert self.napp.liveness_manager.disable.call_count == 1
442	1		assert self.napp.publish_liveness_status.call_count == 1
443
444	1		async def test_endpoint_get_liveness(self):
445			"""Test GET v1/liveness/."""
446	1		self.napp.liveness_manager.enable = MagicMock()
447	1		self.napp.publish_liveness_status = MagicMock()
448	1		url = f"{self.base_endpoint}/liveness/"
449	1		response = await self.api_client.get(url)
450	1		assert response.status_code == 200
451	1		assert response.json() == {"interfaces": []}
452
453	1		async def test_endpoint_get_pair_liveness(self):
454			"""Test GET v1/liveness//pair."""
455	1		self.napp.liveness_manager.enable = MagicMock()
456	1		self.napp.publish_liveness_status = MagicMock()
457	1		url = f"{self.base_endpoint}/liveness/pair"
458	1		response = await self.api_client.get(url)
459	1		assert response.status_code == 200
460			assert response.json() == {"pairs": []}
461

kytos-ng / of_lldp

Pull Request — master (#78)

TestMain.teardown_method() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like