unit.test_main.TestMain.test_bytes_count() - Code Metrics - Inspection of "Merge pull request #9 from kytos-ng/fix/div-by-zer..." - kytos-ng/kytos_stats - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ba1563...974c0c )

by Vinicius

created 2023-09-27 13:37 UTC

unit.test_main.TestMain.test_bytes_count() A

↳ Parent: unit.test_main

Complexity

Conditions

Size

Total Lines	14
Code Lines	12

Duplication

Lines	14
Ratio	100 %

Code Coverage

Tests	12
CRAP Score	1

Importance

Changes

Metric	Value
eloc	12
dl	14
loc	14
rs	9.8
c	0
b	0
f	0
ccs	12
cts	12
cp	1
cc	1
nop	2
crap	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.kytos_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/kytos_stats/v1"

    def test_get_event_listeners(self):
        """Verify all event listeners registered."""
        expected_events = [
            'kytos/of_core.flow_stats.received',
            'kytos/of_core.table_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.kytos_stats.main.Main.flow_from_id")

    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.kytos_stats.main.Main.flow_from_id")

    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.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    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.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    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.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    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.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    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.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    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.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    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.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    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.kytos_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

    @patch("napps.amlight.kytos_stats.main.Main.table_stats_by_dpid_table_id")

    async def test_table_stats_by_dpid_table_id(self, mock_from_table):
        """Test table_stats rest call."""
        table_info = {
            "table_id": 10,
            "active_count": 20,
            "lookup_count": 30,
            "matched_count": 32768
            }
        table_stats_dict_mock = {'10': table_info}
        table_by_sw = {"00:00:00:00:00:00:00:01": table_stats_dict_mock}
        mock_from_table.return_value = table_by_sw

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

    @patch("napps.amlight.kytos_stats.main.Main.table_stats_by_dpid_table_id")

    async def test_table_stats_by_dpid_table_id_without_dpid(self,
                                                             mock_from_table):
        """Test table_stats rest call."""
        table_info = {
            "table_id": 10,
            "active_count": 20,
            "lookup_count": 30,
            "matched_count": 32768
            }
        table_stats_dict_mock = {'10': table_info}
        table_by_sw = {"00:00:00:00:00:00:00:01": table_stats_dict_mock}
        mock_from_table.return_value = table_by_sw

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

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

    @patch("napps.amlight.kytos_stats.main.Main.table_stats_by_dpid_table_id")
    async def test_table_stats_by_dpid_table_id_not_found(self,
                                                          mock_from_table):
        """Test table_stats rest call."""
        table_by_sw = {}
        mock_from_table.return_value = table_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_multipart_replies_tables(self):
        """Helper method to create mock multipart replies tables"""
        table = MagicMock()
        table.table_id = 10
        table.active_count = 0
        table.lookup_count = 0
        table.matched_count = 0

        replies_tables = [table]
        return replies_tables

    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.kytos_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.kytos_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

    @patch("napps.amlight.kytos_stats.main.Main.handle_table_stats_received")
    def test_handle_table_stats_received(self, mock_handle_stats):
        """Test handle_table_stats_received function."""

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

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

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

    def test_handle_table_stats_reply_received(self):
        """Test handle_table_stats_reply_received call."""

        tables_mock = self._get_mocked_multipart_replies_tables()
        self.napp.handle_table_stats_reply_received(tables_mock)
        table = list(self.napp.tables_stats_dict.values())[0]
        assert list(table.keys())[0] == 10

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")
    async def test_flows_counters_div_zero(self, mock_from_flow):
        """Test that there is no error due to division by zero."""
        flow_info = {
            "byte_count": 10,
            "packet_count": 20,
            "duration_sec": 0
            }
        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)
        response = response.json()
        assert response[0]["flow_id"] == flow_id
        assert response[0]["packet_per_second"] == 0

        endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{dpid}"
        response = await self.api_client.get(endpoint)
        response = response.json()
        assert response[0]["flow_id"] == flow_id
        assert response[0]["bits_per_second"] == 0


1			"""Module to test the main napp file."""
2	1		from unittest.mock import MagicMock, patch
3	1		from kytos.lib.helpers import (
4			get_controller_mock,
5			get_test_client,
6			get_kytos_event_mock,
7			get_switch_mock,
8			)
9	1		from napps.amlight.kytos_stats.main import Main
10
11
12			# pylint: disable=too-many-public-methods, too-many-lines
13	1		class TestMain:
14			"""Test the Main class."""
15
16	1		def setup_method(self):
17			"""Execute steps before each tests."""
18	1		controller = get_controller_mock()
19	1		self.napp = Main(controller)
20	1		self.api_client = get_test_client(controller, self.napp)
21	1		self.base_endpoint = "amlight/kytos_stats/v1"
22
23	1		def test_get_event_listeners(self):
24			"""Verify all event listeners registered."""
25	1		expected_events = [
26			'kytos/of_core.flow_stats.received',
27			'kytos/of_core.table_stats.received'
28			]
29	1		actual_events = self.napp.listeners()
30
31	1		for _event in expected_events:
32	1		assert _event in actual_events
33
34	1		def test_execute(self):
35			"""Test execute."""
36
37	1		def test_shutdown(self):
38			"""Test shutdown."""
39
40	1		def test_flow_from_id(self):
41			"""Test flow_from_id function"""
42	1		flow = self._get_mocked_flow_base()
43	1		self.napp.flows_stats_dict = {
44			flow.id: flow
45			}
46	1		results = self.napp.flow_from_id(flow.id)
47	1		assert results.id == flow.id
48
49	1		def test_flow_from_id__fail(self):
50			"""Test flow_from_id function"""
51	1		flow = self._get_mocked_flow_base()
52	1		self.napp.flows_stats_dict = {
53			flow.id: flow
54			}
55	1		results = self.napp.flow_from_id('1')
56	1		assert results is None
57
58	1		def test_flow_from_id__empty(self):
59			"""Test flow_from_id function when flows_stats_dict is empty"""
60	1		self.napp.flows_stats_dict = {}
61	1		results = self.napp.flow_from_id('1')
62	1		assert results is None
63
64	1		async def test_packet_count_not_found(self):
65			"""Test packet_count rest call with wrong flow_id."""
66	1		flow_id = "123456789"
67	1		endpoint = f"{self.base_endpoint}/packet_count/{flow_id}"
68	1		response = await self.api_client.get(endpoint)
69	1		assert response.status_code == 404
70	1		assert response.json()["description"] == "Flow does not exist"
71
72	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.flow_from_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
73	1		async def test_packet_count(self, mock_from_flow):
74			"""Test packet_count rest call."""
75	1		flow_id = '1'
76	1		mock_from_flow.return_value = self._get_mocked_flow_base()
77
78	1		self._patch_switch_flow(flow_id)
79	1		endpoint = f"{self.base_endpoint}/packet_count/{flow_id}"
80	1		response = await self.api_client.get(endpoint)
81	1		assert response.status_code == 200
82	1		json_response = response.json()
83	1		assert json_response["flow_id"] == flow_id
84	1		assert json_response["packet_counter"] == 40
85	1		assert json_response["packet_per_second"] == 2.0
86
87	1		async def test_bytes_count_not_found(self):
88			"""Test bytes_count rest call with wrong flow_id."""
89	1		flow_id = "123456789"
90	1		endpoint = f"{self.base_endpoint}/bytes_count/{flow_id}"
91	1		response = await self.api_client.get(endpoint)
92	1		assert response.status_code == 404
93	1		assert response.json()["description"] == "Flow does not exist"
94
95	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.flow_from_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
96	1		async def test_bytes_count(self, mock_from_flow):
97			"""Test bytes_count rest call."""
98	1		flow_id = '1'
99	1		mock_from_flow.return_value = self._get_mocked_flow_base()
100	1		self._patch_switch_flow(flow_id)
101
102	1		endpoint = f"{self.base_endpoint}/bytes_count/{flow_id}"
103	1		response = await self.api_client.get(endpoint)
104	1		assert response.status_code == 200
105	1		json_response = response.json()
106	1		assert json_response["flow_id"] == flow_id
107	1		assert json_response["bytes_counter"] == 10
108	1		assert json_response["bits_per_second"] == 4.0
109
110	1		async def test_packet_count_per_flow_empty(self):
111			"""Test packet_count rest call with a flow that does not exist ."""
112	1		flow_id = "123456789"
113	1		endpoint = f"{self.base_endpoint}/packet_count/per_flow/{flow_id}"
114	1		response = await self.api_client.get(endpoint)
115	1		assert response.status_code == 200
116	1		assert len(response.json()) == 0
117
118	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
119	1		async def test_packet_count_per_flow(self, mock_from_flow):
120			"""Test packet_count_per_flow rest call."""
121	1		flow_info = {
122			"byte_count": 10,
123			"duration_sec": 20,
124			"duration_nsec": 30,
125			"packet_count": 40,
126			"cookie": 12310228866111668291,
127			"match": {"in_port": 1},
128			"priority": 32768
129			}
130	1		flow_id = '6055f13593fad45e0b4699f49d56b105'
131	1		flow_stats_dict_mock = {flow_id: flow_info}
132	1		dpid = "00:00:00:00:00:00:00:01"
133	1		flow_by_sw = {dpid: flow_stats_dict_mock}
134	1		mock_from_flow.return_value = flow_by_sw
135
136	1		self._patch_switch_flow(flow_id)
137	1		endpoint = f"{self.base_endpoint}/packet_count/per_flow/{dpid}"
138	1		response = await self.api_client.get(endpoint)
139
140	1		json_response = response.json()
141	1		assert json_response[0]["flow_id"] == flow_id
142	1		assert json_response[0]["packet_counter"] == 40
143	1		assert json_response[0]["packet_per_second"] == 2.0
144
145	1		async def test_bytes_count_per_flow__empty(self):
146			"""Test bytes_count rest call with a flow that does not exist ."""
147	1		flow_id = "123456789"
148	1		endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{flow_id}"
149	1		response = await self.api_client.get(endpoint)
150	1		assert response.status_code == 200
151	1		assert len(response.json()) == 0
152
153	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
154	1		async def test_bytes_count_per_flow(self, mock_from_flow):
155			"""Test bytes_count_per_flow rest call."""
156	1		flow_info = {
157			"byte_count": 10,
158			"duration_sec": 20,
159			"duration_nsec": 30,
160			"packet_count": 40,
161			"cookie": 12310228866111668291,
162			"match": {"in_port": 1},
163			"priority": 32768
164			}
165	1		flow_id = '6055f13593fad45e0b4699f49d56b105'
166	1		flow_stats_dict_mock = {flow_id: flow_info}
167	1		dpid = "00:00:00:00:00:00:00:01"
168	1		flow_by_sw = {dpid: flow_stats_dict_mock}
169	1		mock_from_flow.return_value = flow_by_sw
170
171	1		self._patch_switch_flow(flow_id)
172
173	1		endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{dpid}"
174	1		response = await self.api_client.get(endpoint)
175	1		assert response.status_code == 200
176
177	1		json_response = response.json()
178	1		assert json_response[0]["flow_id"] == flow_id
179	1		assert json_response[0]["bytes_counter"] == 10
180	1		assert json_response[0]["bits_per_second"] == 4.0
181
182	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
183	1		async def test_flows_counters_packet(self, mock_from_flow):
184			"""Test flows_counters function for packet"""
185	1		flow_info = {
186			"byte_count": 10,
187			"duration_sec": 20,
188			"duration_nsec": 30,
189			"packet_count": 40,
190			"cookie": 12310228866111668291,
191			"match": {"in_port": 1},
192			"priority": 32768
193			}
194	1		flow_id = '6055f13593fad45e0b4699f49d56b105'
195	1		flow_stats_dict_mock = {flow_id: flow_info}
196	1		dpid = "00:00:00:00:00:00:00:01"
197	1		flow_by_sw = {dpid: flow_stats_dict_mock}
198	1		mock_from_flow.return_value = flow_by_sw
199
200	1		endpoint = f"{self.base_endpoint}/packet_count/per_flow/{dpid}"
201	1		response = await self.api_client.get(endpoint)
202	1		assert response.status_code == 200
203	1		assert len(response.json()) == 1
204
205	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
206	1		async def test_flows_counters_bytes(self, mock_from_flow):
207			"""Test flows_counters function for bytes"""
208	1		flow_info = {
209			"byte_count": 10,
210			"duration_sec": 20,
211			"duration_nsec": 30,
212			"packet_count": 40,
213			"cookie": 12310228866111668291,
214			"match": {"in_port": 1},
215			"priority": 32768
216			}
217	1		flow_id = '6055f13593fad45e0b4699f49d56b105'
218	1		flow_stats_dict_mock = {flow_id: flow_info}
219	1		dpid = "00:00:00:00:00:00:00:01"
220	1		flow_by_sw = {dpid: flow_stats_dict_mock}
221	1		mock_from_flow.return_value = flow_by_sw
222
223	1		endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{dpid}"
224	1		response = await self.api_client.get(endpoint)
225	1		assert response.status_code == 200
226	1		assert len(response.json()) == 1
227
228	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
229	1		async def test_flow_stats_by_dpid_flow_id(self, mock_from_flow):
230			"""Test flow_stats rest call."""
231	1		flow_info = {
232			"byte_count": 10,
233			"duration_sec": 20,
234			"duration_nsec": 30,
235			"packet_count": 40,
236			"cookie": 12310228866111668291,
237			"match": {"in_port": 1},
238			"priority": 32768
239			}
240	1		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
241	1		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
242	1		mock_from_flow.return_value = flow_by_sw
243
244	1		endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
245	1		url = f"{self.base_endpoint}{endpoint}"
246	1		response = await self.api_client.get(url)
247	1		assert response.status_code == 200
248	1		expected = flow_by_sw
249	1		assert response.json() == expected
250
251	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
252	1		async def test_flow_stats_by_dpid_flow_id_without_dpid(self,
253			mock_from_flow):
254			"""Test flow_stats rest call."""
255	1		flow_info = {
256			"byte_count": 10,
257			"duration_sec": 20,
258			"duration_nsec": 30,
259			"packet_count": 40,
260			"cookie": 12310228866111668291,
261			"match": {"in_port": 1},
262			"priority": 32768
263			}
264	1		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
265	1		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
266	1		mock_from_flow.return_value = flow_by_sw
267
268	1		endpoint = "/flow/stats"
269	1		url = f"{self.base_endpoint}{endpoint}"
270	1		response = await self.api_client.get(url)
271	1		assert response.status_code == 200
272
273	1		expected = flow_by_sw
274	1		assert response.json() == expected
275
276	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
277	1		async def test_flow_stats_by_dpid_flow_id_with_dpid(self, mock_from_flow):
278			"""Test flow_stats rest call."""
279	1		flow_info = {
280			"byte_count": 10,
281			"duration_sec": 20,
282			"duration_nsec": 30,
283			"packet_count": 40,
284			"cookie": 12310228866111668291,
285			"match": {"in_port": 1},
286			"priority": 32768
287			}
288	1		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
289	1		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
290	1		mock_from_flow.return_value = flow_by_sw
291
292	1		endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
293	1		url = f"{self.base_endpoint}{endpoint}"
294	1		response = await self.api_client.get(url)
295	1		assert response.status_code == 200
296
297	1		expected = flow_by_sw
298	1		assert response.json() == expected
299
300	1		@patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")
301	1		async def test_flow_stats_by_dpid_flow_id_not_found(self, mock_from_flow):
302			"""Test flow_stats rest call."""
303	1		flow_by_sw = {}
304	1		mock_from_flow.return_value = flow_by_sw
305	1		endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
306	1		url = f"{self.base_endpoint}{endpoint}"
307	1		response = await self.api_client.get(url)
308	1		assert response.status_code == 200
309	1		assert len(response.json()) == 0
310
311	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.table_stats_by_dpid_table_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
312	1		async def test_table_stats_by_dpid_table_id(self, mock_from_table):
313			"""Test table_stats rest call."""
314	1		table_info = {
315			"table_id": 10,
316			"active_count": 20,
317			"lookup_count": 30,
318			"matched_count": 32768
319			}
320	1		table_stats_dict_mock = {'10': table_info}
321	1		table_by_sw = {"00:00:00:00:00:00:00:01": table_stats_dict_mock}
322	1		mock_from_table.return_value = table_by_sw
323
324	1		endpoint = "/table/stats?dpid=00:00:00:00:00:00:00:01&table=10"
325	1		url = f"{self.base_endpoint}{endpoint}"
326	1		response = await self.api_client.get(url)
327	1		assert response.status_code == 200
328	1		expected = table_by_sw
329	1		assert response.json() == expected
330
331	1	View Code Duplication	@patch("napps.amlight.kytos_stats.main.Main.table_stats_by_dpid_table_id")
			0 ignored issues – show Duplication introduced 2023-05-22 15:35 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
332	1		async def test_table_stats_by_dpid_table_id_without_dpid(self,
333			mock_from_table):
334			"""Test table_stats rest call."""
335	1		table_info = {
336			"table_id": 10,
337			"active_count": 20,
338			"lookup_count": 30,
339			"matched_count": 32768
340			}
341	1		table_stats_dict_mock = {'10': table_info}
342	1		table_by_sw = {"00:00:00:00:00:00:00:01": table_stats_dict_mock}
343	1		mock_from_table.return_value = table_by_sw
344
345	1		endpoint = "/table/stats"
346	1		url = f"{self.base_endpoint}{endpoint}"
347	1		response = await self.api_client.get(url)
348	1		assert response.status_code == 200
349
350	1		expected = table_by_sw
351	1		assert response.json() == expected
352
353	1		@patch("napps.amlight.kytos_stats.main.Main.table_stats_by_dpid_table_id")
354	1		async def test_table_stats_by_dpid_table_id_not_found(self,
355			mock_from_table):
356			"""Test table_stats rest call."""
357	1		table_by_sw = {}
358	1		mock_from_table.return_value = table_by_sw
359	1		endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
360	1		url = f"{self.base_endpoint}{endpoint}"
361	1		response = await self.api_client.get(url)
362	1		assert response.status_code == 200
363	1		assert len(response.json()) == 0
364
365	1		def _patch_switch_flow(self, flow_id):
366			"""Helper method to patch controller to return switch/flow data."""
367			# patching the flow_stats object in the switch
368	1		flow = self._get_mocked_flow_stats()
369	1		flow.id = flow_id
370	1		switch = MagicMock()
371	1		self.napp.controller.switches = {"1": switch}
372	1		self.napp.controller.get_switch_by_dpid = MagicMock()
373	1		self.napp.controller.get_switch_by_dpid.return_value = switch
374
375	1		def _get_mocked_flow_stats(self):
376			"""Helper method to create a mock flow_stats object."""
377	1		flow_stats = MagicMock()
378	1		flow_stats.id = 123
379	1		flow_stats.byte_count = 10
380	1		flow_stats.duration_sec = 20
381	1		flow_stats.duration_nsec = 30
382	1		flow_stats.packet_count = 40
383	1		return flow_stats
384
385	1		def _get_mocked_multipart_replies_flows(self):
386			"""Helper method to create mock multipart replies flows"""
387	1		flow = self._get_mocked_flow_base()
388
389	1		instruction = MagicMock()
390	1		flow.instructions = [instruction]
391
392	1		replies_flows = [flow]
393	1		return replies_flows
394
395	1		def _get_mocked_multipart_replies_tables(self):
396			"""Helper method to create mock multipart replies tables"""
397	1		table = MagicMock()
398	1		table.table_id = 10
399	1		table.active_count = 0
400	1		table.lookup_count = 0
401	1		table.matched_count = 0
402
403	1		replies_tables = [table]
404	1		return replies_tables
405
406	1		def _get_mocked_flow_base(self):
407			"""Helper method to create a mock flow object."""
408	1		flow = MagicMock()
409	1		flow.id = 456
410	1		flow.switch = None
411	1		flow.table_id = None
412	1		flow.match = None
413	1		flow.priority = None
414	1		flow.idle_timeout = None
415	1		flow.hard_timeout = None
416	1		flow.cookie = None
417	1		flow.stats = self._get_mocked_flow_stats()
418	1		return flow
419
420	1		@patch("napps.amlight.kytos_stats.main.Main.handle_stats_reply_received")
421	1		def test_handle_stats_received(self, mock_handle_stats):
422			"""Test handle_stats_received function."""
423
424	1		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
425	1		replies_flows = self._get_mocked_multipart_replies_flows()
426	1		name = "kytos/of_core.flow_stats.received"
427	1		content = {"switch": switch_v0x04, "replies_flows": replies_flows}
428
429	1		event = get_kytos_event_mock(name=name, content=content)
430
431	1		self.napp.handle_stats_received(event)
432	1		mock_handle_stats.assert_called_once()
433
434	1		@patch("napps.amlight.kytos_stats.main.Main.handle_stats_reply_received")
435	1		def test_handle_stats_received__fail(self, mock_handle_stats):
436			"""Test handle_stats_received function for
437			fail when replies_flows is not in content."""
438
439	1		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
440	1		name = "kytos/of_core.flow_stats.received"
441	1		content = {"switch": switch_v0x04}
442
443	1		event = get_kytos_event_mock(name=name, content=content)
444
445	1		self.napp.handle_stats_received(event)
446	1		mock_handle_stats.assert_not_called()
447
448	1		def test_handle_stats_reply_received(self):
449			"""Test handle_stats_reply_received call."""
450
451	1		flows_mock = self._get_mocked_multipart_replies_flows()
452	1		self.napp.handle_stats_reply_received(flows_mock)
453
454	1		assert list(self.napp.flows_stats_dict.values())[0].id == 456
455
456	1		@patch("napps.amlight.kytos_stats.main.Main.handle_table_stats_received")
457	1		def test_handle_table_stats_received(self, mock_handle_stats):
458			"""Test handle_table_stats_received function."""
459
460	1		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
461	1		replies_tables = self._get_mocked_multipart_replies_tables()
462	1		name = "kytos/of_core.table_stats.received"
463	1		content = {"switch": switch_v0x04, "replies_tables": replies_tables}
464
465	1		event = get_kytos_event_mock(name=name, content=content)
466
467	1		self.napp.handle_table_stats_received(event)
468	1		mock_handle_stats.assert_called_once()
469
470	1		def test_handle_table_stats_reply_received(self):
471			"""Test handle_table_stats_reply_received call."""
472
473	1		tables_mock = self._get_mocked_multipart_replies_tables()
474	1		self.napp.handle_table_stats_reply_received(tables_mock)
475	1		table = list(self.napp.tables_stats_dict.values())[0]
476	1		assert list(table.keys())[0] == 10
477
478	1		@patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")
479	1		async def test_flows_counters_div_zero(self, mock_from_flow):
480			"""Test that there is no error due to division by zero."""
481	1		flow_info = {
482			"byte_count": 10,
483			"packet_count": 20,
484			"duration_sec": 0
485			}
486	1		flow_id = '6055f13593fad45e0b4699f49d56b105'
487	1		flow_stats_dict_mock = {flow_id: flow_info}
488	1		dpid = "00:00:00:00:00:00:00:01"
489	1		flow_by_sw = {dpid: flow_stats_dict_mock}
490	1		mock_from_flow.return_value = flow_by_sw
491
492	1		self._patch_switch_flow(flow_id)
493	1		endpoint = f"{self.base_endpoint}/packet_count/per_flow/{dpid}"
494	1		response = await self.api_client.get(endpoint)
495	1		response = response.json()
496	1		assert response[0]["flow_id"] == flow_id
497	1		assert response[0]["packet_per_second"] == 0
498
499	1		endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{dpid}"
500	1		response = await self.api_client.get(endpoint)
501	1		response = response.json()
502	1		assert response[0]["flow_id"] == flow_id
503			assert response[0]["bits_per_second"] == 0
504

kytos-ng / kytos_stats

Push — master ( ba1563...974c0c )

unit.test_main.TestMain.test_bytes_count() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like