build.tests.unit.test_main.TestMain.test_shutdown() - Code Metrics - Inspection of "Merge pull request #44 from kytos-ng/feature/ep032" - kytos-ng/flow_stats - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 0a5ad3...adbf5d )

unknown

created 2022-11-23 17:37 UTC

TestMain.test_shutdown() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	2
Code Lines	1

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	1
dl	0
loc	2
rs	10
c	0
b	0
f	0
cc	1
nop	1

"""Module to test the main napp file."""
import json
from unittest import TestCase
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(TestCase):
    """Test the Main class."""

    def setUp(self):
        """Execute steps before each tests.

        Set the server_name_url_url from amlight/flow_stats
        """
        self.server_name_url = \
            "http://localhost:8181/api/amlight/flow_stats/v1"
        self.napp = Main(get_controller_mock())

    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:
            self.assertIn(_event, actual_events, _event)

    @staticmethod
    def get_napp_urls(napp):
        """Return the amlight/flow_stats urls.

        The urls will be like:

        urls = [
            (options, methods, url)
        ]

        """
        controller = napp.controller
        controller.api_server.register_napp_endpoints(napp)

        urls = []
        for rule in controller.api_server.app.url_map.iter_rules():
            options = {}
            for arg in rule.arguments:
                options[arg] = f"[{0}]".format(arg)

            if f"{napp.username}/{napp.name}" in str(rule):
                urls.append((options, rule.methods, f"{str(rule)}"))

        return urls

    def test_verify_api_urls(self):
        """Verify all APIs registered."""

        expected_urls = [
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/v1/flow/stats/",
            ),
            (
                {"flow_id": "[flow_id]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/v1/packet_count/",
            ),
            (
                {"flow_id": "[flow_id]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/v1/bytes_count/",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/v1/packet_count/per_flow/",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/v1/bytes_count/per_flow/",
            ),
        ]
        urls = self.get_napp_urls(self.napp)
        assert len(expected_urls) == len(urls)

    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

    def test_packet_count__fail(self):
        """Test bytes_count rest call with wrong flow_id."""
        flow_id = "123456789"
        rest_name = "packet_count"
        response = self._get_rest_response(rest_name, flow_id)

        assert response.data == b"Flow does not exist"

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

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

        rest_name = "packet_count"
        self._patch_switch_flow(flow_id)

        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)
        assert json_response["flow_id"] == flow_id
        assert json_response["packet_counter"] == 40
        assert json_response["packet_per_second"] == 2.0

    def test_bytes_count__fail(self):
        """Test bytes_count rest call with wrong flow_id."""
        flow_id = "123456789"
        rest_name = "bytes_count"
        response = self._get_rest_response(rest_name, flow_id)

        assert response.data == b"Flow does not exist"

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

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

        rest_name = "bytes_count"
        self._patch_switch_flow(flow_id)

        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)
        assert json_response["flow_id"] == flow_id
        assert json_response["bytes_counter"] == 10
        assert json_response["bits_per_second"] == 4.0

    def test_packet_count_per_flow__empty(self):
        """Test packet_count rest call with a flow that does not exist ."""
        flow_id = "123456789"
        rest_name = "packet_count/per_flow"
        response = self._get_rest_response(rest_name, flow_id)
        json_response = json.loads(response.data)
        assert len(json_response) == 0

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

    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_id = "00:00:00:00:00:00:00:01"
        flow_by_sw = {dpid_id: flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        rest_name = "packet_count/per_flow"
        self._patch_switch_flow(flow_id)
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)
        assert json_response[0]["flow_id"] == flow_id
        assert json_response[0]["packet_counter"] == 40
        assert json_response[0]["packet_per_second"] == 2.0

    def test_bytes_count_per_flow__empty(self):
        """Test bytes_count rest call with a flow that does not exist ."""
        flow_id = "123456789"
        rest_name = "bytes_count/per_flow"
        response = self._get_rest_response(rest_name, flow_id)
        json_response = json.loads(response.data)
        assert len(json_response) == 0

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

    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_id = "00:00:00:00:00:00:00:01"
        flow_by_sw = {dpid_id: flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        rest_name = "bytes_count/per_flow"
        self._patch_switch_flow(flow_id)

        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)
        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")

    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_id = "00:00:00:00:00:00:00:01"
        flow_by_sw = {dpid_id: flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        with get_controller_mock().api_server.app.app_context():
            response = self.napp.flows_counters('packet_count', dpid_id)
            json_response = response.json
            assert len(json_response) == 1

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

    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_id = "00:00:00:00:00:00:00:01"
        flow_by_sw = {dpid_id: flow_stats_dict_mock}
        mock_from_flow.return_value = flow_by_sw

        with get_controller_mock().api_server.app.app_context():
            response = self.napp.flows_counters('byte_count', dpid_id)
            json_response = response.json
            assert len(json_response) == 1

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

    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

        api = get_test_client(self.napp.controller, self.napp)
        endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
        url = f"{self.server_name_url}"+endpoint

        response = api.get(url)
        expected = flow_by_sw
        assert response.json == expected
        assert response.status_code == 200

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

    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

        api = get_test_client(self.napp.controller, self.napp)
        endpoint = "/flow/stats"
        url = f"{self.server_name_url}"+endpoint

        response = api.get(url)
        expected = flow_by_sw
        assert response.json == expected
        assert response.status_code == 200

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

    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

        api = get_test_client(self.napp.controller, self.napp)
        endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
        url = f"{self.server_name_url}"+endpoint

        response = api.get(url)
        expected = flow_by_sw
        assert response.json == expected
        assert response.status_code == 200

    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
    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

        api = get_test_client(self.napp.controller, self.napp)
        endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
        url = f"{self.server_name_url}"+endpoint

        response = api.get(url)
        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_rest_response(self, rest_name, url_id):
        """Helper method to call a rest endpoint."""
        # call rest
        api = get_test_client(get_controller_mock(), self.napp)
        url = f"{self.server_name_url}/{rest_name}/{url_id}"
        response = api.get(url, content_type="application/json")

        return response

    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		import json
3		from unittest import TestCase
4		from unittest.mock import MagicMock, patch
5		from kytos.lib.helpers import (
6		get_controller_mock,
7		get_test_client,
8		get_kytos_event_mock,
9		get_switch_mock,
10		)
11		from napps.amlight.flow_stats.main import Main
12
13
14		# pylint: disable=too-many-public-methods, too-many-lines
15		class TestMain(TestCase):
16		"""Test the Main class."""
17
18		def setUp(self):
19		"""Execute steps before each tests.
20
21		Set the server_name_url_url from amlight/flow_stats
22		"""
23		self.server_name_url = \
24		"http://localhost:8181/api/amlight/flow_stats/v1"
25		self.napp = Main(get_controller_mock())
26
27		def test_get_event_listeners(self):
28		"""Verify all event listeners registered."""
29		expected_events = [
30		'kytos/of_core.flow_stats.received'
31		]
32		actual_events = self.napp.listeners()
33
34		for _event in expected_events:
35		self.assertIn(_event, actual_events, _event)
36
37		@staticmethod
38		def get_napp_urls(napp):
39		"""Return the amlight/flow_stats urls.
40
41		The urls will be like:
42
43		urls = [
44		(options, methods, url)
45		]
46
47		"""
48		controller = napp.controller
49		controller.api_server.register_napp_endpoints(napp)
50
51		urls = []
52		for rule in controller.api_server.app.url_map.iter_rules():
53		options = {}
54		for arg in rule.arguments:
55		options[arg] = f"[{0}]".format(arg)
56
57		if f"{napp.username}/{napp.name}" in str(rule):
58		urls.append((options, rule.methods, f"{str(rule)}"))
59
60		return urls
61
62		def test_verify_api_urls(self):
63		"""Verify all APIs registered."""
64
65		expected_urls = [
66		(
67		{"dpid": "[dpid]"},
68		{"OPTIONS", "HEAD", "GET"},
69		"/api/amlight/flow_stats/v1/flow/stats/",
70		),
71		(
72		{"flow_id": "[flow_id]"},
73		{"OPTIONS", "HEAD", "GET"},
74		"/api/amlight/flow_stats/v1/packet_count/",
75		),
76		(
77		{"flow_id": "[flow_id]"},
78		{"OPTIONS", "HEAD", "GET"},
79		"/api/amlight/flow_stats/v1/bytes_count/",
80		),
81		(
82		{"dpid": "[dpid]"},
83		{"OPTIONS", "HEAD", "GET"},
84		"/api/amlight/flow_stats/v1/packet_count/per_flow/",
85		),
86		(
87		{"dpid": "[dpid]"},
88		{"OPTIONS", "HEAD", "GET"},
89		"/api/amlight/flow_stats/v1/bytes_count/per_flow/",
90		),
91		]
92		urls = self.get_napp_urls(self.napp)
93		assert len(expected_urls) == len(urls)
94
95		def test_execute(self):
96		"""Test execute."""
97
98		def test_shutdown(self):
99		"""Test shutdown."""
100
101		def test_flow_from_id(self):
102		"""Test flow_from_id function"""
103		flow = self._get_mocked_flow_base()
104		self.napp.flows_stats_dict = {
105		flow.id: flow
106		}
107		results = self.napp.flow_from_id(flow.id)
108		assert results.id == flow.id
109
110		def test_flow_from_id__fail(self):
111		"""Test flow_from_id function"""
112		flow = self._get_mocked_flow_base()
113		self.napp.flows_stats_dict = {
114		flow.id: flow
115		}
116		results = self.napp.flow_from_id('1')
117		assert results is None
118
119		def test_flow_from_id__empty(self):
120		"""Test flow_from_id function when flows_stats_dict is empty"""
121		self.napp.flows_stats_dict = {}
122		results = self.napp.flow_from_id('1')
123		assert results is None
124
125		def test_packet_count__fail(self):
126		"""Test bytes_count rest call with wrong flow_id."""
127		flow_id = "123456789"
128		rest_name = "packet_count"
129		response = self._get_rest_response(rest_name, flow_id)
130
131		assert response.data == b"Flow does not exist"
132
133	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...
134		def test_packet_count(self, mock_from_flow):
135		"""Test packet_count rest call."""
136		flow_id = '1'
137		dpid_id = '1'
138		mock_from_flow.return_value = self._get_mocked_flow_base()
139
140		rest_name = "packet_count"
141		self._patch_switch_flow(flow_id)
142
143		response = self._get_rest_response(rest_name, dpid_id)
144		json_response = json.loads(response.data)
145		assert json_response["flow_id"] == flow_id
146		assert json_response["packet_counter"] == 40
147		assert json_response["packet_per_second"] == 2.0
148
149		def test_bytes_count__fail(self):
150		"""Test bytes_count rest call with wrong flow_id."""
151		flow_id = "123456789"
152		rest_name = "bytes_count"
153		response = self._get_rest_response(rest_name, flow_id)
154
155		assert response.data == b"Flow does not exist"
156
157	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...
158		def test_bytes_count(self, mock_from_flow):
159		"""Test bytes_count rest call."""
160		flow_id = '1'
161		dpid_id = '1'
162		mock_from_flow.return_value = self._get_mocked_flow_base()
163
164		rest_name = "bytes_count"
165		self._patch_switch_flow(flow_id)
166
167		response = self._get_rest_response(rest_name, dpid_id)
168		json_response = json.loads(response.data)
169		assert json_response["flow_id"] == flow_id
170		assert json_response["bytes_counter"] == 10
171		assert json_response["bits_per_second"] == 4.0
172
173		def test_packet_count_per_flow__empty(self):
174		"""Test packet_count rest call with a flow that does not exist ."""
175		flow_id = "123456789"
176		rest_name = "packet_count/per_flow"
177		response = self._get_rest_response(rest_name, flow_id)
178		json_response = json.loads(response.data)
179		assert len(json_response) == 0
180
181	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		def test_packet_count_per_flow(self, mock_from_flow):
183		"""Test packet_count_per_flow rest call."""
184		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		flow_id = '6055f13593fad45e0b4699f49d56b105'
194		flow_stats_dict_mock = {flow_id: flow_info}
195		dpid_id = "00:00:00:00:00:00:00:01"
196		flow_by_sw = {dpid_id: flow_stats_dict_mock}
197		mock_from_flow.return_value = flow_by_sw
198
199		rest_name = "packet_count/per_flow"
200		self._patch_switch_flow(flow_id)
201		response = self._get_rest_response(rest_name, dpid_id)
202		json_response = json.loads(response.data)
203		assert json_response[0]["flow_id"] == flow_id
204		assert json_response[0]["packet_counter"] == 40
205		assert json_response[0]["packet_per_second"] == 2.0
206
207		def test_bytes_count_per_flow__empty(self):
208		"""Test bytes_count rest call with a flow that does not exist ."""
209		flow_id = "123456789"
210		rest_name = "bytes_count/per_flow"
211		response = self._get_rest_response(rest_name, flow_id)
212		json_response = json.loads(response.data)
213		assert len(json_response) == 0
214
215	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...
216		def test_bytes_count_per_flow(self, mock_from_flow):
217		"""Test bytes_count_per_flow rest call."""
218		flow_info = {
219		"byte_count": 10,
220		"duration_sec": 20,
221		"duration_nsec": 30,
222		"packet_count": 40,
223		"cookie": 12310228866111668291,
224		"match": {"in_port": 1},
225		"priority": 32768
226		}
227		flow_id = '6055f13593fad45e0b4699f49d56b105'
228		flow_stats_dict_mock = {flow_id: flow_info}
229		dpid_id = "00:00:00:00:00:00:00:01"
230		flow_by_sw = {dpid_id: flow_stats_dict_mock}
231		mock_from_flow.return_value = flow_by_sw
232
233		rest_name = "bytes_count/per_flow"
234		self._patch_switch_flow(flow_id)
235
236		response = self._get_rest_response(rest_name, dpid_id)
237		json_response = json.loads(response.data)
238		assert json_response[0]["flow_id"] == flow_id
239		assert json_response[0]["bytes_counter"] == 10
240		assert json_response[0]["bits_per_second"] == 4.0
241
242	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...
243		def test_flows_counters_packet(self, mock_from_flow):
244		"""Test flows_counters function for packet"""
245		flow_info = {
246		"byte_count": 10,
247		"duration_sec": 20,
248		"duration_nsec": 30,
249		"packet_count": 40,
250		"cookie": 12310228866111668291,
251		"match": {"in_port": 1},
252		"priority": 32768
253		}
254		flow_id = '6055f13593fad45e0b4699f49d56b105'
255		flow_stats_dict_mock = {flow_id: flow_info}
256		dpid_id = "00:00:00:00:00:00:00:01"
257		flow_by_sw = {dpid_id: flow_stats_dict_mock}
258		mock_from_flow.return_value = flow_by_sw
259
260		with get_controller_mock().api_server.app.app_context():
261		response = self.napp.flows_counters('packet_count', dpid_id)
262		json_response = response.json
263		assert len(json_response) == 1
264
265	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...
266		def test_flows_counters_bytes(self, mock_from_flow):
267		"""Test flows_counters function for bytes"""
268		flow_info = {
269		"byte_count": 10,
270		"duration_sec": 20,
271		"duration_nsec": 30,
272		"packet_count": 40,
273		"cookie": 12310228866111668291,
274		"match": {"in_port": 1},
275		"priority": 32768
276		}
277		flow_id = '6055f13593fad45e0b4699f49d56b105'
278		flow_stats_dict_mock = {flow_id: flow_info}
279		dpid_id = "00:00:00:00:00:00:00:01"
280		flow_by_sw = {dpid_id: flow_stats_dict_mock}
281		mock_from_flow.return_value = flow_by_sw
282
283		with get_controller_mock().api_server.app.app_context():
284		response = self.napp.flows_counters('byte_count', dpid_id)
285		json_response = response.json
286		assert len(json_response) == 1
287
288	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...
289		def test_flow_stats_by_dpid_flow_id(self, mock_from_flow):
290		"""Test flow_stats rest call."""
291		flow_info = {
292		"byte_count": 10,
293		"duration_sec": 20,
294		"duration_nsec": 30,
295		"packet_count": 40,
296		"cookie": 12310228866111668291,
297		"match": {"in_port": 1},
298		"priority": 32768
299		}
300		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
301		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
302		mock_from_flow.return_value = flow_by_sw
303
304		api = get_test_client(self.napp.controller, self.napp)
305		endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
306		url = f"{self.server_name_url}"+endpoint
307
308		response = api.get(url)
309		expected = flow_by_sw
310		assert response.json == expected
311		assert response.status_code == 200
312
313	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...
314		def test_flow_stats_by_dpid_flow_id_without_dpid(self, mock_from_flow):
315		"""Test flow_stats rest call."""
316		flow_info = {
317		"byte_count": 10,
318		"duration_sec": 20,
319		"duration_nsec": 30,
320		"packet_count": 40,
321		"cookie": 12310228866111668291,
322		"match": {"in_port": 1},
323		"priority": 32768
324		}
325		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
326		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
327		mock_from_flow.return_value = flow_by_sw
328
329		api = get_test_client(self.napp.controller, self.napp)
330		endpoint = "/flow/stats"
331		url = f"{self.server_name_url}"+endpoint
332
333		response = api.get(url)
334		expected = flow_by_sw
335		assert response.json == expected
336		assert response.status_code == 200
337
338	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...
339		def test_flow_stats_by_dpid_flow_id_with_dpid(self, mock_from_flow):
340		"""Test flow_stats rest call."""
341		flow_info = {
342		"byte_count": 10,
343		"duration_sec": 20,
344		"duration_nsec": 30,
345		"packet_count": 40,
346		"cookie": 12310228866111668291,
347		"match": {"in_port": 1},
348		"priority": 32768
349		}
350		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}
351		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
352		mock_from_flow.return_value = flow_by_sw
353
354		api = get_test_client(self.napp.controller, self.napp)
355		endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
356		url = f"{self.server_name_url}"+endpoint
357
358		response = api.get(url)
359		expected = flow_by_sw
360		assert response.json == expected
361		assert response.status_code == 200
362
363		@patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
364		def test_flow_stats_by_dpid_flow_id_not_found(self, mock_from_flow):
365		"""Test flow_stats rest call."""
366		flow_by_sw = {}
367		mock_from_flow.return_value = flow_by_sw
368
369		api = get_test_client(self.napp.controller, self.napp)
370		endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
371		url = f"{self.server_name_url}"+endpoint
372
373		response = api.get(url)
374		assert len(response.json) == 0
375
376		def _patch_switch_flow(self, flow_id):
377		"""Helper method to patch controller to return switch/flow data."""
378		# patching the flow_stats object in the switch
379		flow = self._get_mocked_flow_stats()
380		flow.id = flow_id
381		switch = MagicMock()
382		self.napp.controller.switches = {"1": switch}
383		self.napp.controller.get_switch_by_dpid = MagicMock()
384		self.napp.controller.get_switch_by_dpid.return_value = switch
385
386		def _get_rest_response(self, rest_name, url_id):
387		"""Helper method to call a rest endpoint."""
388		# call rest
389		api = get_test_client(get_controller_mock(), self.napp)
390		url = f"{self.server_name_url}/{rest_name}/{url_id}"
391		response = api.get(url, content_type="application/json")
392
393		return response
394
395		def _get_mocked_flow_stats(self):
396		"""Helper method to create a mock flow_stats object."""
397		flow_stats = MagicMock()
398		flow_stats.id = 123
399		flow_stats.byte_count = 10
400		flow_stats.duration_sec = 20
401		flow_stats.duration_nsec = 30
402		flow_stats.packet_count = 40
403		return flow_stats
404
405		def _get_mocked_multipart_replies_flows(self):
406		"""Helper method to create mock multipart replies flows"""
407		flow = self._get_mocked_flow_base()
408
409		instruction = MagicMock()
410		flow.instructions = [instruction]
411
412		replies_flows = [flow]
413		return replies_flows
414
415		def _get_mocked_flow_base(self):
416		"""Helper method to create a mock flow object."""
417		flow = MagicMock()
418		flow.id = 456
419		flow.switch = None
420		flow.table_id = None
421		flow.match = None
422		flow.priority = None
423		flow.idle_timeout = None
424		flow.hard_timeout = None
425		flow.cookie = None
426		flow.stats = self._get_mocked_flow_stats()
427		return flow
428
429		@patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
430		def test_handle_stats_received(self, mock_handle_stats):
431		"""Test handle_stats_received function."""
432
433		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
434		replies_flows = self._get_mocked_multipart_replies_flows()
435		name = "kytos/of_core.flow_stats.received"
436		content = {"switch": switch_v0x04, "replies_flows": replies_flows}
437
438		event = get_kytos_event_mock(name=name, content=content)
439
440		self.napp.handle_stats_received(event)
441		mock_handle_stats.assert_called_once()
442
443		@patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
444		def test_handle_stats_received__fail(self, mock_handle_stats):
445		"""Test handle_stats_received function for
446		fail when replies_flows is not in content."""
447
448		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
449		name = "kytos/of_core.flow_stats.received"
450		content = {"switch": switch_v0x04}
451
452		event = get_kytos_event_mock(name=name, content=content)
453
454		self.napp.handle_stats_received(event)
455		mock_handle_stats.assert_not_called()
456
457		def test_handle_stats_reply_received(self):
458		"""Test handle_stats_reply_received call."""
459
460		flows_mock = self._get_mocked_multipart_replies_flows()
461		self.napp.handle_stats_reply_received(flows_mock)
462
463		assert list(self.napp.flows_stats_dict.values())[0].id == 456
464

kytos-ng / flow_stats

Push — master ( 0a5ad3...adbf5d )

TestMain.test_shutdown() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like