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

Passed

Push — master ( 412579...0a5ad3 )

unknown

created 2022-11-10 16:36 UTC

TestMain.test_flow_match() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	27
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	20
dl	0
loc	27
rs	9.4
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/v1"
        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/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_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_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)
        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_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)

        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_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

        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_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

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

kytos-ng / flow_stats

Push — master ( 412579...0a5ad3 )

TestMain.test_flow_match() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like