build.tests.unit.test_main.TestMain._patch_switch_flow() - Code Metrics - Inspection of "Remove GenericFlow && Adapt GET flow/stats/<dpid>" - kytos-ng/flow_stats - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#42)

unknown

created 2022-10-25 19:42 UTC

TestMain._patch_switch_flow() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	9
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

"""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"
        self.napp = Main(get_controller_mock())

    @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/flow/stats/",
            ),
            (
                {"flow_id": "[flow_id]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/packet_count/",
            ),
            (
                {"flow_id": "[flow_id]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/bytes_count/",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/packet_count/per_flow/",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/packet_count/sum/",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/bytes_count/per_flow/",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/bytes_count/sum/",
            ),
        ]
        urls = self.get_napp_urls(self.napp)
        assert len(expected_urls) == len(urls)

    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

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

    def test_packet_count_per_flow(self, mock_from_flow):
        """Test packet_count_per_flow rest call."""
        flow_stats = {
            'byte_count': 10,
            'duration_sec': 20,
            'duration_nsec': 30,
            'packet_count': 40
            }
        flow_id = '6055f13593fad45e0b4699f49d56b105'
        flow_stats_dict_mock = {flow_id: flow_stats}
        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)

        mock_from_flow.return_value = flow_by_sw
        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

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

    def test_packet_count_sum(self, mock_from_flow):
        """Test packet_count_sum rest call."""
        flow_stats = {
            'byte_count': 10,
            'duration_sec': 20,
            'duration_nsec': 30,
            'packet_count': 40
            }
        flow_id = '6055f13593fad45e0b4699f49d56b105'
        flow_stats_dict_mock = {flow_id: flow_stats}
        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/sum"
        self._patch_switch_flow(flow_id)

        mock_from_flow.return_value = flow_by_sw
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)
        assert json_response == 40

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

    def test_bytes_count_sum(self, mock_from_flow):
        """Test bytes_count_sum rest call."""
        flow_stats = {
            'byte_count': 10,
            'duration_sec': 20,
            'duration_nsec': 30,
            'packet_count': 40
            }
        flow_id = '6055f13593fad45e0b4699f49d56b105'
        flow_stats_dict_mock = {flow_id: flow_stats}
        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/sum"
        self._patch_switch_flow(flow_id)

        mock_from_flow.return_value = flow_by_sw
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)
        assert json_response == 10

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

    def test_bytes_count_per_flow(self, mock_from_flow):
        """Test bytes_count_per_flow rest call."""
        flow_stats = {
            'byte_count': 10,
            'duration_sec': 20,
            'duration_nsec': 30,
            'packet_count': 40
            }
        flow_id = '6055f13593fad45e0b4699f49d56b105'
        flow_stats_dict_mock = {flow_id: flow_stats}
        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)

        mock_from_flow.return_value = flow_by_sw
        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")

    def test_flow_stats(self, mock_from_flow):
        """Test flow_stats rest call."""
        flow_stats = {
            'byte_count': 148,
            'duration_sec': 1589,
            'duration_nsec': 556000000,
            'packet_count': 2
            }
        flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_stats}
        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)
        url = f"{self.server_name_url}/flow/stats?dpid=00:00:00:00:00:00:00:01"

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

    def test_flow_stats_with_dpid(self, mock_from_flow):
        """Test flow_stats rest call."""
        flow_stats = {
            'byte_count': 148,
            'duration_sec': 1589,
            'duration_nsec': 556000000,
            'packet_count': 2
            }
        flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_stats}
        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)
        url = f"{self.server_name_url}/flow/stats?dpid=00:00:00:00:00:00:00:01"

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

    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 = "http://localhost:8181/api/amlight/flow_stats"
24		self.napp = Main(get_controller_mock())
25
26		@staticmethod
27		def get_napp_urls(napp):
28		"""Return the amlight/flow_stats urls.
29
30		The urls will be like:
31
32		urls = [
33		(options, methods, url)
34		]
35
36		"""
37		controller = napp.controller
38		controller.api_server.register_napp_endpoints(napp)
39
40		urls = []
41		for rule in controller.api_server.app.url_map.iter_rules():
42		options = {}
43		for arg in rule.arguments:
44		options[arg] = f"[{0}]".format(arg)
45
46		if f"{napp.username}/{napp.name}" in str(rule):
47		urls.append((options, rule.methods, f"{str(rule)}"))
48
49		return urls
50
51		def test_verify_api_urls(self):
52		"""Verify all APIs registered."""
53
54		expected_urls = [
55		(
56		{"dpid": "[dpid]"},
57		{"OPTIONS", "HEAD", "GET"},
58		"/api/amlight/flow_stats/flow/stats/",
59		),
60		(
61		{"flow_id": "[flow_id]"},
62		{"OPTIONS", "HEAD", "GET"},
63		"/api/amlight/flow_stats/packet_count/",
64		),
65		(
66		{"flow_id": "[flow_id]"},
67		{"OPTIONS", "HEAD", "GET"},
68		"/api/amlight/flow_stats/bytes_count/",
69		),
70		(
71		{"dpid": "[dpid]"},
72		{"OPTIONS", "HEAD", "GET"},
73		"/api/amlight/flow_stats/packet_count/per_flow/",
74		),
75		(
76		{"dpid": "[dpid]"},
77		{"OPTIONS", "HEAD", "GET"},
78		"/api/amlight/flow_stats/packet_count/sum/",
79		),
80		(
81		{"dpid": "[dpid]"},
82		{"OPTIONS", "HEAD", "GET"},
83		"/api/amlight/flow_stats/bytes_count/per_flow/",
84		),
85		(
86		{"dpid": "[dpid]"},
87		{"OPTIONS", "HEAD", "GET"},
88		"/api/amlight/flow_stats/bytes_count/sum/",
89		),
90		]
91		urls = self.get_napp_urls(self.napp)
92		assert len(expected_urls) == len(urls)
93
94		def test_packet_count__fail(self):
95		"""Test bytes_count rest call with wrong flow_id."""
96		flow_id = "123456789"
97		rest_name = "packet_count"
98		response = self._get_rest_response(rest_name, flow_id)
99
100		assert response.data == b"Flow does not exist"
101
102	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...
103		def test_packet_count(self, mock_from_flow):
104		"""Test packet_count rest call."""
105		flow_id = '1'
106		dpid_id = '1'
107		mock_from_flow.return_value = self._get_mocked_flow_base()
108
109		rest_name = "packet_count"
110		self._patch_switch_flow(flow_id)
111
112		response = self._get_rest_response(rest_name, dpid_id)
113		json_response = json.loads(response.data)
114		assert json_response["flow_id"] == flow_id
115		assert json_response["packet_counter"] == 40
116		assert json_response["packet_per_second"] == 2.0
117
118		def test_bytes_count__fail(self):
119		"""Test bytes_count rest call with wrong flow_id."""
120		flow_id = "123456789"
121		rest_name = "bytes_count"
122		response = self._get_rest_response(rest_name, flow_id)
123
124		assert response.data == b"Flow does not exist"
125
126	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...
127		def test_bytes_count(self, mock_from_flow):
128		"""Test bytes_count rest call."""
129		flow_id = '1'
130		dpid_id = '1'
131		mock_from_flow.return_value = self._get_mocked_flow_base()
132
133		rest_name = "bytes_count"
134		self._patch_switch_flow(flow_id)
135
136		response = self._get_rest_response(rest_name, dpid_id)
137		json_response = json.loads(response.data)
138		assert json_response["flow_id"] == flow_id
139		assert json_response["bytes_counter"] == 10
140		assert json_response["bits_per_second"] == 4.0
141
142	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main._flow_stats")
		0 ignored issues – show Duplication introduced 2022-10-25 19:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
143		def test_packet_count_per_flow(self, mock_from_flow):
144		"""Test packet_count_per_flow rest call."""
145		flow_stats = {
146		'byte_count': 10,
147		'duration_sec': 20,
148		'duration_nsec': 30,
149		'packet_count': 40
150		}
151		flow_id = '6055f13593fad45e0b4699f49d56b105'
152		flow_stats_dict_mock = {flow_id: flow_stats}
153		dpid_id = "00:00:00:00:00:00:00:01"
154		flow_by_sw = {dpid_id: flow_stats_dict_mock}
155		mock_from_flow.return_value = flow_by_sw
156
157		rest_name = "packet_count/per_flow"
158		self._patch_switch_flow(flow_id)
159
160		mock_from_flow.return_value = flow_by_sw
161		response = self._get_rest_response(rest_name, dpid_id)
162		json_response = json.loads(response.data)
163		assert json_response[0]["flow_id"] == flow_id
164		assert json_response[0]["packet_counter"] == 40
165		assert json_response[0]["packet_per_second"] == 2.0
166
167	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main._flow_stats")
		0 ignored issues – show Duplication introduced 2022-10-25 19:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
168		def test_packet_count_sum(self, mock_from_flow):
169		"""Test packet_count_sum rest call."""
170		flow_stats = {
171		'byte_count': 10,
172		'duration_sec': 20,
173		'duration_nsec': 30,
174		'packet_count': 40
175		}
176		flow_id = '6055f13593fad45e0b4699f49d56b105'
177		flow_stats_dict_mock = {flow_id: flow_stats}
178		dpid_id = "00:00:00:00:00:00:00:01"
179		flow_by_sw = {dpid_id: flow_stats_dict_mock}
180		mock_from_flow.return_value = flow_by_sw
181
182		rest_name = "packet_count/sum"
183		self._patch_switch_flow(flow_id)
184
185		mock_from_flow.return_value = flow_by_sw
186		response = self._get_rest_response(rest_name, dpid_id)
187		json_response = json.loads(response.data)
188		assert json_response == 40
189
190	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main._flow_stats")
		0 ignored issues – show Duplication introduced 2022-10-25 19:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
191		def test_bytes_count_sum(self, mock_from_flow):
192		"""Test bytes_count_sum rest call."""
193		flow_stats = {
194		'byte_count': 10,
195		'duration_sec': 20,
196		'duration_nsec': 30,
197		'packet_count': 40
198		}
199		flow_id = '6055f13593fad45e0b4699f49d56b105'
200		flow_stats_dict_mock = {flow_id: flow_stats}
201		dpid_id = "00:00:00:00:00:00:00:01"
202		flow_by_sw = {dpid_id: flow_stats_dict_mock}
203		mock_from_flow.return_value = flow_by_sw
204
205		rest_name = "bytes_count/sum"
206		self._patch_switch_flow(flow_id)
207
208		mock_from_flow.return_value = flow_by_sw
209		response = self._get_rest_response(rest_name, dpid_id)
210		json_response = json.loads(response.data)
211		assert json_response == 10
212
213	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main._flow_stats")
		0 ignored issues – show Duplication introduced 2022-10-25 19:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
214		def test_bytes_count_per_flow(self, mock_from_flow):
215		"""Test bytes_count_per_flow rest call."""
216		flow_stats = {
217		'byte_count': 10,
218		'duration_sec': 20,
219		'duration_nsec': 30,
220		'packet_count': 40
221		}
222		flow_id = '6055f13593fad45e0b4699f49d56b105'
223		flow_stats_dict_mock = {flow_id: flow_stats}
224		dpid_id = "00:00:00:00:00:00:00:01"
225		flow_by_sw = {dpid_id: flow_stats_dict_mock}
226		mock_from_flow.return_value = flow_by_sw
227
228		rest_name = "bytes_count/per_flow"
229		self._patch_switch_flow(flow_id)
230
231		mock_from_flow.return_value = flow_by_sw
232		response = self._get_rest_response(rest_name, dpid_id)
233		json_response = json.loads(response.data)
234		assert json_response[0]["flow_id"] == flow_id
235		assert json_response[0]["bytes_counter"] == 10
236		assert json_response[0]["bits_per_second"] == 4.0
237
238	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main._flow_stats")
		0 ignored issues – show Duplication introduced 2022-10-25 19:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
239		def test_flow_stats(self, mock_from_flow):
240		"""Test flow_stats rest call."""
241		flow_stats = {
242		'byte_count': 148,
243		'duration_sec': 1589,
244		'duration_nsec': 556000000,
245		'packet_count': 2
246		}
247		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_stats}
248		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
249		mock_from_flow.return_value = flow_by_sw
250
251		api = get_test_client(self.napp.controller, self.napp)
252		url = f"{self.server_name_url}/flow/stats?dpid=00:00:00:00:00:00:00:01"
253
254		response = api.get(url)
255		expected = flow_by_sw
256		assert response.json == expected
257		assert response.status_code == 200
258
259	View Code Duplication	@patch("napps.amlight.flow_stats.main.Main._flow_stats")
		0 ignored issues – show Duplication introduced 2022-10-25 19:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
260		def test_flow_stats_with_dpid(self, mock_from_flow):
261		"""Test flow_stats rest call."""
262		flow_stats = {
263		'byte_count': 148,
264		'duration_sec': 1589,
265		'duration_nsec': 556000000,
266		'packet_count': 2
267		}
268		flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_stats}
269		flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
270		mock_from_flow.return_value = flow_by_sw
271
272		api = get_test_client(self.napp.controller, self.napp)
273		url = f"{self.server_name_url}/flow/stats?dpid=00:00:00:00:00:00:00:01"
274
275		response = api.get(url)
276		expected = flow_by_sw
277		assert response.json == expected
278		assert response.status_code == 200
279
280		def _patch_switch_flow(self, flow_id):
281		"""Helper method to patch controller to return switch/flow data."""
282		# patching the flow_stats object in the switch
283		flow = self._get_mocked_flow_stats()
284		flow.id = flow_id
285		switch = MagicMock()
286		self.napp.controller.switches = {"1": switch}
287		self.napp.controller.get_switch_by_dpid = MagicMock()
288		self.napp.controller.get_switch_by_dpid.return_value = switch
289
290		def _get_rest_response(self, rest_name, url_id):
291		"""Helper method to call a rest endpoint."""
292		# call rest
293		api = get_test_client(get_controller_mock(), self.napp)
294		url = f"{self.server_name_url}/{rest_name}/{url_id}"
295		response = api.get(url, content_type="application/json")
296
297		return response
298
299		def _get_mocked_flow_stats(self):
300		"""Helper method to create a mock flow_stats object."""
301		flow_stats = MagicMock()
302		flow_stats.id = 123
303		flow_stats.byte_count = 10
304		flow_stats.duration_sec = 20
305		flow_stats.duration_nsec = 30
306		flow_stats.packet_count = 40
307		return flow_stats
308
309		def _get_mocked_multipart_replies_flows(self):
310		"""Helper method to create mock multipart replies flows"""
311		flow = self._get_mocked_flow_base()
312
313		instruction = MagicMock()
314		flow.instructions = [instruction]
315
316		replies_flows = [flow]
317		return replies_flows
318
319		def _get_mocked_flow_base(self):
320		"""Helper method to create a mock flow object."""
321		flow = MagicMock()
322		flow.id = 456
323		flow.switch = None
324		flow.table_id = None
325		flow.match = None
326		flow.priority = None
327		flow.idle_timeout = None
328		flow.hard_timeout = None
329		flow.cookie = None
330		flow.stats = self._get_mocked_flow_stats()
331		return flow
332
333		@patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
334		def test_handle_stats_received(self, mock_handle_stats):
335		"""Test handle_stats_received function."""
336
337		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
338		replies_flows = self._get_mocked_multipart_replies_flows()
339		name = "kytos/of_core.flow_stats.received"
340		content = {"switch": switch_v0x04, "replies_flows": replies_flows}
341
342		event = get_kytos_event_mock(name=name, content=content)
343
344		self.napp.handle_stats_received(event)
345		mock_handle_stats.assert_called_once()
346
347		@patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
348		def test_handle_stats_received_fail(self, mock_handle_stats):
349		"""Test handle_stats_received function for
350		fail when replies_flows is not in content."""
351
352		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
353		name = "kytos/of_core.flow_stats.received"
354		content = {"switch": switch_v0x04}
355
356		event = get_kytos_event_mock(name=name, content=content)
357
358		self.napp.handle_stats_received(event)
359		mock_handle_stats.assert_not_called()
360
361		def test_handle_stats_reply_received(self):
362		"""Test handle_stats_reply_received call."""
363
364		flows_mock = self._get_mocked_multipart_replies_flows()
365		self.napp.handle_stats_reply_received(flows_mock)
366
367		assert list(self.napp.flows_stats_dict.values())[0].id == 456
368

kytos-ng / flow_stats

Pull Request — master (#42)

TestMain._patch_switch_flow() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like