unit.test_main

Complexity

Total Complexity

29

Size/Duplication

Total Lines	389
Duplicated Lines	50.13 %

Test Coverage

Coverage

100%

Importance

Changes

0

28 Methods

Rating	Name	Duplication	Size	Complexity
A	TestMain.test_handle_stats_received()	0	13	1
A	TestMain._get_mocked_multipart_replies_flows()	0	9	1
A	TestMain.test_packet_count_not_found()	0	7	1
A	TestMain.setup_method()	0	6	1
A	TestMain.test_flows_counters_packet()	22	22	1
A	TestMain.test_execute()	0	2	1
A	TestMain.test_flow_stats_by_dpid_flow_id_with_dpid()	23	23	1
A	TestMain.test_bytes_count()	14	14	1
A	TestMain.test_bytes_count_per_flow__empty()	0	7	1
A	TestMain._get_mocked_flow_stats()	0	9	1
A	TestMain.test_bytes_count_not_found()	0	7	1
A	TestMain.test_packet_count_per_flow()	26	26	1
A	TestMain.test_flow_from_id()	0	8	1
A	TestMain.test_flow_stats_by_dpid_flow_id()	22	22	1
A	TestMain._patch_switch_flow()	0	9	1
A	TestMain.test_flow_from_id__empty()	0	5	1
A	TestMain.test_handle_stats_reply_received()	0	7	1
A	TestMain.test_flows_counters_bytes()	22	22	1
A	TestMain.test_packet_count_per_flow_empty()	0	7	1
A	TestMain.test_flow_stats_by_dpid_flow_id_not_found()	0	10	1
A	TestMain.test_bytes_count_per_flow()	28	28	1
A	TestMain.test_handle_stats_received__fail()	0	13	1
A	TestMain.test_shutdown()	0	2	1
A	TestMain.test_flow_from_id__fail()	0	8	1
A	TestMain._get_mocked_flow_base()	0	13	1
A	TestMain.test_flow_stats_by_dpid_flow_id_without_dpid()	24	24	1
A	TestMain.test_get_event_listeners()	0	9	2
A	TestMain.test_packet_count()	14	14	1

"""Module to test the main napp file."""
from unittest.mock import MagicMock, patch
from kytos.lib.helpers import (
    get_controller_mock,
    get_test_client,
    get_kytos_event_mock,
    get_switch_mock,
)
from napps.amlight.flow_stats.main import Main


# pylint: disable=too-many-public-methods, too-many-lines
class TestMain:
    """Test the Main class."""

    def setup_method(self):
        """Execute steps before each tests."""
        controller = get_controller_mock()
        self.napp = Main(controller)
        self.api_client = get_test_client(controller, self.napp)
        self.base_endpoint = "amlight/flow_stats/v1"

    def test_get_event_listeners(self):
        """Verify all event listeners registered."""
        expected_events = [
            'kytos/of_core.flow_stats.received'
        ]
        actual_events = self.napp.listeners()

        for _event in expected_events:
            assert _event in actual_events

    def test_execute(self):
        """Test execute."""

    def test_shutdown(self):
        """Test shutdown."""

    def test_flow_from_id(self):
        """Test flow_from_id function"""
        flow = self._get_mocked_flow_base()
        self.napp.flows_stats_dict = {
            flow.id: flow
        }
        results = self.napp.flow_from_id(flow.id)
        assert results.id == flow.id

    def test_flow_from_id__fail(self):
        """Test flow_from_id function"""
        flow = self._get_mocked_flow_base()
        self.napp.flows_stats_dict = {
            flow.id: flow
        }
        results = self.napp.flow_from_id('1')
        assert results is None

    def test_flow_from_id__empty(self):
        """Test flow_from_id function when flows_stats_dict is empty"""
        self.napp.flows_stats_dict = {}
        results = self.napp.flow_from_id('1')
        assert results is None

    async def test_packet_count_not_found(self):
        """Test packet_count rest call with wrong flow_id."""
        flow_id = "123456789"
        endpoint = f"{self.base_endpoint}/packet_count/{flow_id}"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 404
        assert response.json()["description"] == "Flow does not exist"

    @patch("napps.amlight.flow_stats.main.Main.flow_from_id")

This code seems to be duplicated in your project.

    async def test_packet_count(self, mock_from_flow):
        """Test packet_count rest call."""
        flow_id = '1'
        mock_from_flow.return_value = self._get_mocked_flow_base()

        self._patch_switch_flow(flow_id)
        endpoint = f"{self.base_endpoint}/packet_count/{flow_id}"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 200
        json_response = response.json()
        assert json_response["flow_id"] == flow_id
        assert json_response["packet_counter"] == 40
        assert json_response["packet_per_second"] == 2.0

    async def test_bytes_count_not_found(self):
        """Test bytes_count rest call with wrong flow_id."""
        flow_id = "123456789"
        endpoint = f"{self.base_endpoint}/bytes_count/{flow_id}"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 404
        assert response.json()["description"] == "Flow does not exist"

    @patch("napps.amlight.flow_stats.main.Main.flow_from_id")

This code seems to be duplicated in your project.

    async def test_bytes_count(self, mock_from_flow):
        """Test bytes_count rest call."""
        flow_id = '1'
        mock_from_flow.return_value = self._get_mocked_flow_base()
        self._patch_switch_flow(flow_id)

        endpoint = f"{self.base_endpoint}/bytes_count/{flow_id}"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 200
        json_response = response.json()
        assert json_response["flow_id"] == flow_id
        assert json_response["bytes_counter"] == 10
        assert json_response["bits_per_second"] == 4.0

    async def test_packet_count_per_flow_empty(self):
        """Test packet_count rest call with a flow that does not exist ."""
        flow_id = "123456789"
        endpoint = f"{self.base_endpoint}/packet_count/per_flow/{flow_id}"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 200
        assert len(response.json()) == 0

    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")

This code seems to be duplicated in your project.

    async def test_packet_count_per_flow(self, mock_from_flow):
        """Test packet_count_per_flow rest call."""
        flow_info = {
            "byte_count": 10,
            "duration_sec": 20,
            "duration_nsec": 30,
            "packet_count": 40,
            "cookie": 12310228866111668291,
            "match": {"in_port": 1},
            "priority": 32768
            }
        flow_id = '6055f13593fad45e0b4699f49d56b105'
        flow_stats_dict_mock = {flow_id: flow_info}
        dpid = "00:00:00:00:00:00:00:01"
        flow_by_sw = {dpid: flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        self._patch_switch_flow(flow_id)
        endpoint = f"{self.base_endpoint}/packet_count/per_flow/{dpid}"
        response = await self.api_client.get(endpoint)

        json_response = response.json()
        assert json_response[0]["flow_id"] == flow_id
        assert json_response[0]["packet_counter"] == 40
        assert json_response[0]["packet_per_second"] == 2.0

    async def test_bytes_count_per_flow__empty(self):
        """Test bytes_count rest call with a flow that does not exist ."""
        flow_id = "123456789"
        endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{flow_id}"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 200
        assert len(response.json()) == 0

    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")

This code seems to be duplicated in your project.

    async def test_bytes_count_per_flow(self, mock_from_flow):
        """Test bytes_count_per_flow rest call."""
        flow_info = {
            "byte_count": 10,
            "duration_sec": 20,
            "duration_nsec": 30,
            "packet_count": 40,
            "cookie": 12310228866111668291,
            "match": {"in_port": 1},
            "priority": 32768
            }
        flow_id = '6055f13593fad45e0b4699f49d56b105'
        flow_stats_dict_mock = {flow_id: flow_info}
        dpid = "00:00:00:00:00:00:00:01"
        flow_by_sw = {dpid: flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        self._patch_switch_flow(flow_id)

        endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{dpid}"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 200

        json_response = response.json()
        assert json_response[0]["flow_id"] == flow_id
        assert json_response[0]["bytes_counter"] == 10
        assert json_response[0]["bits_per_second"] == 4.0

    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")

This code seems to be duplicated in your project.

    async def test_flows_counters_packet(self, mock_from_flow):
        """Test flows_counters function for packet"""
        flow_info = {
            "byte_count": 10,
            "duration_sec": 20,
            "duration_nsec": 30,
            "packet_count": 40,
            "cookie": 12310228866111668291,
            "match": {"in_port": 1},
            "priority": 32768
            }
        flow_id = '6055f13593fad45e0b4699f49d56b105'
        flow_stats_dict_mock = {flow_id: flow_info}
        dpid = "00:00:00:00:00:00:00:01"
        flow_by_sw = {dpid: flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        endpoint = f"{self.base_endpoint}/packet_count/per_flow/{dpid}"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 200
        assert len(response.json()) == 1

    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")

This code seems to be duplicated in your project.

    async def test_flows_counters_bytes(self, mock_from_flow):
        """Test flows_counters function for bytes"""
        flow_info = {
            "byte_count": 10,
            "duration_sec": 20,
            "duration_nsec": 30,
            "packet_count": 40,
            "cookie": 12310228866111668291,
            "match": {"in_port": 1},
            "priority": 32768
            }
        flow_id = '6055f13593fad45e0b4699f49d56b105'
        flow_stats_dict_mock = {flow_id: flow_info}
        dpid = "00:00:00:00:00:00:00:01"
        flow_by_sw = {dpid: flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{dpid}"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 200
        assert len(response.json()) == 1

    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")

This code seems to be duplicated in your project.

    async def test_flow_stats_by_dpid_flow_id(self, mock_from_flow):
        """Test flow_stats rest call."""
        flow_info = {
            "byte_count": 10,
            "duration_sec": 20,
            "duration_nsec": 30,
            "packet_count": 40,
            "cookie": 12310228866111668291,
            "match": {"in_port": 1},
            "priority": 32768
            }
        flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
        flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
        url = f"{self.base_endpoint}{endpoint}"
        response = await self.api_client.get(url)
        assert response.status_code == 200
        expected = flow_by_sw
        assert response.json() == expected

    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")

This code seems to be duplicated in your project.

    async def test_flow_stats_by_dpid_flow_id_without_dpid(self,
                                                           mock_from_flow):
        """Test flow_stats rest call."""
        flow_info = {
            "byte_count": 10,
            "duration_sec": 20,
            "duration_nsec": 30,
            "packet_count": 40,
            "cookie": 12310228866111668291,
            "match": {"in_port": 1},
            "priority": 32768
            }
        flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
        flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        endpoint = "/flow/stats"
        url = f"{self.base_endpoint}{endpoint}"
        response = await self.api_client.get(url)
        assert response.status_code == 200

        expected = flow_by_sw
        assert response.json() == expected

    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")

This code seems to be duplicated in your project.

    async def test_flow_stats_by_dpid_flow_id_with_dpid(self, mock_from_flow):
        """Test flow_stats rest call."""
        flow_info = {
            "byte_count": 10,
            "duration_sec": 20,
            "duration_nsec": 30,
            "packet_count": 40,
            "cookie": 12310228866111668291,
            "match": {"in_port": 1},
            "priority": 32768
            }
        flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
        flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
        url = f"{self.base_endpoint}{endpoint}"
        response = await self.api_client.get(url)
        assert response.status_code == 200

        expected = flow_by_sw
        assert response.json() == expected

    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
    async def test_flow_stats_by_dpid_flow_id_not_found(self, mock_from_flow):
        """Test flow_stats rest call."""
        flow_by_sw = {}
        mock_from_flow.return_value = flow_by_sw
        endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
        url = f"{self.base_endpoint}{endpoint}"
        response = await self.api_client.get(url)
        assert response.status_code == 200
        assert len(response.json()) == 0

    def _patch_switch_flow(self, flow_id):
        """Helper method to patch controller to return switch/flow data."""
        # patching the flow_stats object in the switch
        flow = self._get_mocked_flow_stats()
        flow.id = flow_id
        switch = MagicMock()
        self.napp.controller.switches = {"1": switch}
        self.napp.controller.get_switch_by_dpid = MagicMock()
        self.napp.controller.get_switch_by_dpid.return_value = switch

    def _get_mocked_flow_stats(self):
        """Helper method to create a mock flow_stats object."""
        flow_stats = MagicMock()
        flow_stats.id = 123
        flow_stats.byte_count = 10
        flow_stats.duration_sec = 20
        flow_stats.duration_nsec = 30
        flow_stats.packet_count = 40
        return flow_stats

    def _get_mocked_multipart_replies_flows(self):
        """Helper method to create mock multipart replies flows"""
        flow = self._get_mocked_flow_base()

        instruction = MagicMock()
        flow.instructions = [instruction]

        replies_flows = [flow]
        return replies_flows

    def _get_mocked_flow_base(self):
        """Helper method to create a mock flow object."""
        flow = MagicMock()
        flow.id = 456
        flow.switch = None
        flow.table_id = None
        flow.match = None
        flow.priority = None
        flow.idle_timeout = None
        flow.hard_timeout = None
        flow.cookie = None
        flow.stats = self._get_mocked_flow_stats()
        return flow

    @patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
    def test_handle_stats_received(self, mock_handle_stats):
        """Test handle_stats_received function."""

        switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        replies_flows = self._get_mocked_multipart_replies_flows()
        name = "kytos/of_core.flow_stats.received"
        content = {"switch": switch_v0x04, "replies_flows": replies_flows}

        event = get_kytos_event_mock(name=name, content=content)

        self.napp.handle_stats_received(event)
        mock_handle_stats.assert_called_once()

    @patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
    def test_handle_stats_received__fail(self, mock_handle_stats):
        """Test handle_stats_received function for
        fail when replies_flows is not in content."""

        switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        name = "kytos/of_core.flow_stats.received"
        content = {"switch": switch_v0x04}

        event = get_kytos_event_mock(name=name, content=content)

        self.napp.handle_stats_received(event)
        mock_handle_stats.assert_not_called()

    def test_handle_stats_reply_received(self):
        """Test handle_stats_reply_received call."""

        flows_mock = self._get_mocked_multipart_replies_flows()
        self.napp.handle_stats_reply_received(flows_mock)

        assert list(self.napp.flows_stats_dict.values())[0].id == 456