build.tests.unit.test_main.TestMain.test_bytes_count_fail() - Code Metrics - Inspection of "Drop GenericFlow abstraction" - kytos-ng/flow_stats - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#38)

unknown

created 2022-09-22 19:14 UTC

TestMain.test_bytes_count_fail() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	7
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	7
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
from napps.kytos.of_core.v0x04.flow import Action as Action40
from napps.kytos.of_core.v0x04.flow import Flow as Flow40
from napps.kytos.of_core.v0x04.match_fields import MatchFieldFactory
from napps.kytos.of_core.v0x04.flow import Match as Match40
from pyof.foundation.basic_types import UBInt32


# 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/match/ ",
            ),
            (
                {"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)
        self.assertEqual(len(expected_urls), len(urls))

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

        self.assertEqual(response.data, b"Flow does not exist")

    def test_packet_count(self):
        """Test packet_count rest call."""
        flow_id = "1"
        rest_name = "packet_count"
        self._patch_switch_flow(flow_id)
        response = self._get_rest_response(rest_name, flow_id)

        json_response = json.loads(response.data)
        self.assertEqual(json_response["flow_id"], flow_id)
        self.assertEqual(json_response["packet_counter"], 40)
        self.assertEqual(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)

        self.assertEqual(response.data, b"Flow does not exist")

    def test_bytes_count(self):
        """Test bytes_count rest call."""
        flow_id = "1"
        rest_name = "bytes_count"
        self._patch_switch_flow(flow_id)
        response = self._get_rest_response(rest_name, flow_id)

        json_response = json.loads(response.data)
        self.assertEqual(json_response["flow_id"], flow_id)
        self.assertEqual(json_response["bytes_counter"], 10)
        self.assertEqual(json_response["bits_per_second"], 4.0)

    def test_packet_count_per_flow(self):

        """Test packet_count_per_flow rest call."""
        flow_id = "1"
        rest_name = "packet_count/per_flow"
        self._patch_switch_flow(flow_id)

        dpid_id = 111
        response = self._get_rest_response(rest_name, dpid_id)

        json_response = json.loads(response.data)
        self.assertEqual(json_response[0]["flow_id"], flow_id)
        self.assertEqual(json_response[0]["packet_counter"], 40)
        self.assertEqual(json_response[0]["packet_per_second"], 2.0)

    def test_packet_count_sum(self):
        """Test packet_count_sum rest call."""
        flow_id = "1"
        rest_name = "packet_count/sum"
        self._patch_switch_flow(flow_id)

        dpid_id = 111
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)

        self.assertEqual(json_response, 40)

    def test_bytes_count_per_flow(self):

        """Test bytes_count_per_flow rest call."""
        flow_id = "1"
        rest_name = "bytes_count/per_flow"
        self._patch_switch_flow(flow_id)

        dpid_id = 111
        response = self._get_rest_response(rest_name, dpid_id)

        json_response = json.loads(response.data)
        self.assertEqual(json_response[0]["flow_id"], flow_id)
        self.assertEqual(json_response[0]["bytes_counter"], 10)
        self.assertEqual(json_response[0]["bits_per_second"], 4.0)

    def test_bytes_count_sum(self):
        """Test bytes_count_sum rest call."""
        flow_id = "1"
        rest_name = "bytes_count/sum"
        self._patch_switch_flow(flow_id)

        dpid_id = 111
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)

        self.assertEqual(json_response, 10)

    @patch("napps.amlight.flow_stats.main.Main.match_flows")
    def test_flow_match(self, mock_match_flows):
        """Test flow_match rest call."""
        flow = Flow40()
        flow.actions = [
            Action40.from_dict(
                {
                    "action_type": "output",
                    "port": "1",
                }
            ),
        ]
        flow.version = "0x04"
        mock_match_flows.return_value = flow

        flow_id = "1"
        rest_name = "flow/match"
        self._patch_switch_flow(flow_id)

        dpid_id = "aa:00:00:00:00:00:00:11"
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)

        self.assertEqual(response.status_code, 200)
        self.assertEqual(json_response["actions"][0]["action_type"], "output")
        self.assertEqual(json_response["actions"][0]["port"], "1")
        self.assertEqual(json_response["version"], "0x04")

    @patch("napps.amlight.flow_stats.main.Main.match_flows")
    def test_flow_match_fail(self, mock_match_flows):
        """Test flow_match rest call."""
        mock_match_flows.return_value = None

        flow_id = "1"
        rest_name = "flow/match"
        self._patch_switch_flow(flow_id)

        dpid_id = "aa:00:00:00:00:00:00:11"
        response = self._get_rest_response(rest_name, dpid_id)

        self.assertEqual(response.status_code, 404)

    @patch("napps.amlight.flow_stats.main.Main.match_flows")
    def test_flow_stats(self, mock_match_flows):
        """Test flow_match rest call."""
        flow = Flow40()
        flow.actions = [
            Action40.from_dict(
                {
                    "action_type": "output",
                    "port": "1",
                }
            ),
        ]
        flow.version = "0x04"
        mock_match_flows.return_value = [flow]

        flow_id = "1"
        rest_name = "flow/stats"
        self._patch_switch_flow(flow_id)

        dpid_id = "aa:00:00:00:00:00:00:11"
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)

        self.assertEqual(response.status_code, 200)
        print(json_response)
        self.assertEqual(json_response[0]["actions"][0]["action_type"],
                         "output")
        self.assertEqual(json_response[0]["actions"][0]["port"], "1")
        self.assertEqual(json_response[0]["version"], "0x04")

    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()
        switch.flows = [flow]
        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, mock_from_flow):
        """Test handle_stats_reply_received call."""
        event_switch = MagicMock()
        flows_mock = self._get_mocked_multipart_replies_flows()
        self.napp.handle_stats_reply_received(event_switch, flows_mock)

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

kytos-ng / flow_stats

Pull Request — master (#38)

TestMain.test_bytes_count_fail() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like