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

Passed

Pull Request — master (#55)

by Vinicius

created 2023-05-03 17:54 UTC

TestMain.test_bytes_count_not_found() A

↳ Parent: unit.test_main

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	1

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	7
rs	10
c	0
b	0
f	0
ccs	6
cts	6
cp	1
cc	1
nop	1
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.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")

    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")

    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")

    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")

    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")

    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")

    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")

    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")

    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")

    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


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.flow_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/flow_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			]
28	1		actual_events = self.napp.listeners()
29
30	1		for _event in expected_events:
31	1		assert _event in actual_events
32
33	1		def test_execute(self):
34			"""Test execute."""
35
36	1		def test_shutdown(self):
37			"""Test shutdown."""
38
39	1		def test_flow_from_id(self):
40			"""Test flow_from_id function"""
41	1		flow = self._get_mocked_flow_base()
42	1		self.napp.flows_stats_dict = {
43			flow.id: flow
44			}
45	1		results = self.napp.flow_from_id(flow.id)
46	1		assert results.id == flow.id
47
48	1		def test_flow_from_id__fail(self):
49			"""Test flow_from_id function"""
50	1		flow = self._get_mocked_flow_base()
51	1		self.napp.flows_stats_dict = {
52			flow.id: flow
53			}
54	1		results = self.napp.flow_from_id('1')
55	1		assert results is None
56
57	1		def test_flow_from_id__empty(self):
58			"""Test flow_from_id function when flows_stats_dict is empty"""
59	1		self.napp.flows_stats_dict = {}
60	1		results = self.napp.flow_from_id('1')
61	1		assert results is None
62
63	1		async def test_packet_count_not_found(self):
64			"""Test packet_count rest call with wrong flow_id."""
65	1		flow_id = "123456789"
66	1		endpoint = f"{self.base_endpoint}/packet_count/{flow_id}"
67	1		response = await self.api_client.get(endpoint)
68	1		assert response.status_code == 404
69	1		assert response.json()["description"] == "Flow does not exist"
70
71	1	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main.flow_from_id")
			0 ignored issues – show Duplication introduced 2022-08-05 19:38 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
72	1		async def test_packet_count(self, mock_from_flow):
73			"""Test packet_count rest call."""
74	1		flow_id = '1'
75	1		mock_from_flow.return_value = self._get_mocked_flow_base()
76
77	1		self._patch_switch_flow(flow_id)
78	1		endpoint = f"{self.base_endpoint}/packet_count/{flow_id}"
79	1		response = await self.api_client.get(endpoint)
80	1		assert response.status_code == 200
81	1		json_response = response.json()
82	1		assert json_response["flow_id"] == flow_id
83	1		assert json_response["packet_counter"] == 40
84	1		assert json_response["packet_per_second"] == 2.0
85
86	1		async def test_bytes_count_not_found(self):
87			"""Test bytes_count rest call with wrong flow_id."""
88	1		flow_id = "123456789"
89	1		endpoint = f"{self.base_endpoint}/bytes_count/{flow_id}"
90	1		response = await self.api_client.get(endpoint)
91	1		assert response.status_code == 404
92	1		assert response.json()["description"] == "Flow does not exist"
93
94	1	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main.flow_from_id")
			0 ignored issues – show Duplication introduced 2022-08-05 19:38 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
95	1		async def test_bytes_count(self, mock_from_flow):
96			"""Test bytes_count rest call."""
97	1		flow_id = '1'
98	1		mock_from_flow.return_value = self._get_mocked_flow_base()
99	1		self._patch_switch_flow(flow_id)
100
101	1		endpoint = f"{self.base_endpoint}/bytes_count/{flow_id}"
102	1		response = await self.api_client.get(endpoint)
103	1		assert response.status_code == 200
104	1		json_response = response.json()
105	1		assert json_response["flow_id"] == flow_id
106	1		assert json_response["bytes_counter"] == 10
107	1		assert json_response["bits_per_second"] == 4.0
108
109	1		async def test_packet_count_per_flow_empty(self):
110			"""Test packet_count rest call with a flow that does not exist ."""
111	1		flow_id = "123456789"
112	1		endpoint = f"{self.base_endpoint}/packet_count/per_flow/{flow_id}"
113	1		response = await self.api_client.get(endpoint)
114	1		assert response.status_code == 200
115	1		assert len(response.json()) == 0
116
117	1	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2022-10-31 14:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
118	1		async def test_packet_count_per_flow(self, mock_from_flow):
119			"""Test packet_count_per_flow rest call."""
120	1		flow_info = {
121			"byte_count": 10,
122			"duration_sec": 20,
123			"duration_nsec": 30,
124			"packet_count": 40,
125			"cookie": 12310228866111668291,
126			"match": {"in_port": 1},
127			"priority": 32768
128			}
129	1		flow_id = '6055f13593fad45e0b4699f49d56b105'
130	1		flow_stats_dict_mock = {flow_id: flow_info}
131	1		dpid = "00:00:00:00:00:00:00:01"
132	1		flow_by_sw = {dpid: flow_stats_dict_mock}
133	1		mock_from_flow.return_value = flow_by_sw
134
135	1		self._patch_switch_flow(flow_id)
136	1		endpoint = f"{self.base_endpoint}/packet_count/per_flow/{dpid}"
137	1		response = await self.api_client.get(endpoint)
138
139	1		json_response = response.json()
140	1		assert json_response[0]["flow_id"] == flow_id
141	1		assert json_response[0]["packet_counter"] == 40
142	1		assert json_response[0]["packet_per_second"] == 2.0
143
144	1		async def test_bytes_count_per_flow__empty(self):
145			"""Test bytes_count rest call with a flow that does not exist ."""
146	1		flow_id = "123456789"
147	1		endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{flow_id}"
148	1		response = await self.api_client.get(endpoint)
149	1		assert response.status_code == 200
150	1		assert len(response.json()) == 0
151
152	1	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2022-10-31 14:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
153	1		async def test_bytes_count_per_flow(self, mock_from_flow):
154			"""Test bytes_count_per_flow rest call."""
155	1		flow_info = {
156			"byte_count": 10,
157			"duration_sec": 20,
158			"duration_nsec": 30,
159			"packet_count": 40,
160			"cookie": 12310228866111668291,
161			"match": {"in_port": 1},
162			"priority": 32768
163			}
164	1		flow_id = '6055f13593fad45e0b4699f49d56b105'
165	1		flow_stats_dict_mock = {flow_id: flow_info}
166	1		dpid = "00:00:00:00:00:00:00:01"
167	1		flow_by_sw = {dpid: flow_stats_dict_mock}
168	1		mock_from_flow.return_value = flow_by_sw
169
170	1		self._patch_switch_flow(flow_id)
171
172	1		endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{dpid}"
173	1		response = await self.api_client.get(endpoint)
174	1		assert response.status_code == 200
175
176	1		json_response = response.json()
177	1		assert json_response[0]["flow_id"] == flow_id
178	1		assert json_response[0]["bytes_counter"] == 10
179	1		assert json_response[0]["bits_per_second"] == 4.0
180
181	1	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2022-10-31 14:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
182	1		async def test_flows_counters_packet(self, mock_from_flow):
183			"""Test flows_counters function for packet"""
184	1		flow_info = {
185			"byte_count": 10,
186			"duration_sec": 20,
187			"duration_nsec": 30,
188			"packet_count": 40,
189			"cookie": 12310228866111668291,
190			"match": {"in_port": 1},
191			"priority": 32768
192			}
193	1		flow_id = '6055f13593fad45e0b4699f49d56b105'
194	1		flow_stats_dict_mock = {flow_id: flow_info}
195	1		dpid = "00:00:00:00:00:00:00:01"
196	1		flow_by_sw = {dpid: flow_stats_dict_mock}
197	1		mock_from_flow.return_value = flow_by_sw
198
199	1		endpoint = f"{self.base_endpoint}/packet_count/per_flow/{dpid}"
200	1		response = await self.api_client.get(endpoint)
201	1		assert response.status_code == 200
202	1		assert len(response.json()) == 1
203
204	1	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2022-10-31 14:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
205	1		async def test_flows_counters_bytes(self, mock_from_flow):
206			"""Test flows_counters function for bytes"""
207	1		flow_info = {
208			"byte_count": 10,
209			"duration_sec": 20,
210			"duration_nsec": 30,
211			"packet_count": 40,
212			"cookie": 12310228866111668291,
213			"match": {"in_port": 1},
214			"priority": 32768
215			}
216	1		flow_id = '6055f13593fad45e0b4699f49d56b105'
217	1		flow_stats_dict_mock = {flow_id: flow_info}
218	1		dpid = "00:00:00:00:00:00:00:01"
219	1		flow_by_sw = {dpid: flow_stats_dict_mock}
220	1		mock_from_flow.return_value = flow_by_sw
221
222	1		endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{dpid}"
223	1		response = await self.api_client.get(endpoint)
224	1		assert response.status_code == 200
225	1		assert len(response.json()) == 1
226
227	1	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2022-10-31 14:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
228	1		async def test_flow_stats_by_dpid_flow_id(self, mock_from_flow):
229			"""Test flow_stats rest call."""
230	1		flow_info = {
231			"byte_count": 10,
232			"duration_sec": 20,
233			"duration_nsec": 30,
234			"packet_count": 40,
235			"cookie": 12310228866111668291,
236			"match": {"in_port": 1},
237			"priority": 32768
238			}
239	1		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
240	1		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
241	1		mock_from_flow.return_value = flow_by_sw
242
243	1		endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
244	1		url = f"{self.base_endpoint}{endpoint}"
245	1		response = await self.api_client.get(url)
246	1		assert response.status_code == 200
247	1		expected = flow_by_sw
248	1		assert response.json() == expected
249
250	1	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2022-10-31 14:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
251	1		async def test_flow_stats_by_dpid_flow_id_without_dpid(self,
252			mock_from_flow):
253			"""Test flow_stats rest call."""
254	1		flow_info = {
255			"byte_count": 10,
256			"duration_sec": 20,
257			"duration_nsec": 30,
258			"packet_count": 40,
259			"cookie": 12310228866111668291,
260			"match": {"in_port": 1},
261			"priority": 32768
262			}
263	1		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
264	1		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
265	1		mock_from_flow.return_value = flow_by_sw
266
267	1		endpoint = "/flow/stats"
268	1		url = f"{self.base_endpoint}{endpoint}"
269	1		response = await self.api_client.get(url)
270	1		assert response.status_code == 200
271
272	1		expected = flow_by_sw
273	1		assert response.json() == expected
274
275	1	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
			0 ignored issues – show Duplication introduced 2022-11-22 21:40 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
276	1		async def test_flow_stats_by_dpid_flow_id_with_dpid(self, mock_from_flow):
277			"""Test flow_stats rest call."""
278	1		flow_info = {
279			"byte_count": 10,
280			"duration_sec": 20,
281			"duration_nsec": 30,
282			"packet_count": 40,
283			"cookie": 12310228866111668291,
284			"match": {"in_port": 1},
285			"priority": 32768
286			}
287	1		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
288	1		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
289	1		mock_from_flow.return_value = flow_by_sw
290
291	1		endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
292	1		url = f"{self.base_endpoint}{endpoint}"
293	1		response = await self.api_client.get(url)
294	1		assert response.status_code == 200
295
296	1		expected = flow_by_sw
297	1		assert response.json() == expected
298
299	1		@patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
300	1		async def test_flow_stats_by_dpid_flow_id_not_found(self, mock_from_flow):
301			"""Test flow_stats rest call."""
302	1		flow_by_sw = {}
303	1		mock_from_flow.return_value = flow_by_sw
304	1		endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
305	1		url = f"{self.base_endpoint}{endpoint}"
306	1		response = await self.api_client.get(url)
307	1		assert response.status_code == 200
308	1		assert len(response.json()) == 0
309
310	1		def _patch_switch_flow(self, flow_id):
311			"""Helper method to patch controller to return switch/flow data."""
312			# patching the flow_stats object in the switch
313	1		flow = self._get_mocked_flow_stats()
314	1		flow.id = flow_id
315	1		switch = MagicMock()
316	1		self.napp.controller.switches = {"1": switch}
317	1		self.napp.controller.get_switch_by_dpid = MagicMock()
318	1		self.napp.controller.get_switch_by_dpid.return_value = switch
319
320	1		def _get_mocked_flow_stats(self):
321			"""Helper method to create a mock flow_stats object."""
322	1		flow_stats = MagicMock()
323	1		flow_stats.id = 123
324	1		flow_stats.byte_count = 10
325	1		flow_stats.duration_sec = 20
326	1		flow_stats.duration_nsec = 30
327	1		flow_stats.packet_count = 40
328	1		return flow_stats
329
330	1		def _get_mocked_multipart_replies_flows(self):
331			"""Helper method to create mock multipart replies flows"""
332	1		flow = self._get_mocked_flow_base()
333
334	1		instruction = MagicMock()
335	1		flow.instructions = [instruction]
336
337	1		replies_flows = [flow]
338	1		return replies_flows
339
340	1		def _get_mocked_flow_base(self):
341			"""Helper method to create a mock flow object."""
342	1		flow = MagicMock()
343	1		flow.id = 456
344	1		flow.switch = None
345	1		flow.table_id = None
346	1		flow.match = None
347	1		flow.priority = None
348	1		flow.idle_timeout = None
349	1		flow.hard_timeout = None
350	1		flow.cookie = None
351	1		flow.stats = self._get_mocked_flow_stats()
352	1		return flow
353
354	1		@patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
355	1		def test_handle_stats_received(self, mock_handle_stats):
356			"""Test handle_stats_received function."""
357
358	1		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
359	1		replies_flows = self._get_mocked_multipart_replies_flows()
360	1		name = "kytos/of_core.flow_stats.received"
361	1		content = {"switch": switch_v0x04, "replies_flows": replies_flows}
362
363	1		event = get_kytos_event_mock(name=name, content=content)
364
365	1		self.napp.handle_stats_received(event)
366	1		mock_handle_stats.assert_called_once()
367
368	1		@patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
369	1		def test_handle_stats_received__fail(self, mock_handle_stats):
370			"""Test handle_stats_received function for
371			fail when replies_flows is not in content."""
372
373	1		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
374	1		name = "kytos/of_core.flow_stats.received"
375	1		content = {"switch": switch_v0x04}
376
377	1		event = get_kytos_event_mock(name=name, content=content)
378
379	1		self.napp.handle_stats_received(event)
380	1		mock_handle_stats.assert_not_called()
381
382	1		def test_handle_stats_reply_received(self):
383			"""Test handle_stats_reply_received call."""
384
385	1		flows_mock = self._get_mocked_multipart_replies_flows()
386	1		self.napp.handle_stats_reply_received(flows_mock)
387
388			assert list(self.napp.flows_stats_dict.values())[0].id == 456
389

kytos-ng / flow_stats

Pull Request — master (#55)

TestMain.test_bytes_count_not_found() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like