Passed
Pull Request — master (#42)
by
unknown
02:39
created

TestMain.test_bytes_count__fail()   A

Complexity

Conditions 1

Size

Total Lines 7
Code Lines 5

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 5
dl 0
loc 7
rs 10
c 0
b 0
f 0
cc 1
nop 1
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_packet_count__fail(self):
86
        """Test bytes_count rest call with wrong flow_id."""
87
        flow_id = "123456789"
88
        rest_name = "packet_count"
89
        response = self._get_rest_response(rest_name, flow_id)
90
91
        assert response.data == b"Flow does not exist"
92
93 View Code Duplication
    @patch("napps.amlight.flow_stats.main.Main.flow_from_id")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
94
    def test_packet_count(self, mock_from_flow):
95
        """Test packet_count rest call."""
96
        flow_id = '1'
97
        dpid_id = '1'
98
        mock_from_flow.return_value = self._get_mocked_flow_base()
99
100
        rest_name = "packet_count"
101
        self._patch_switch_flow(flow_id)
102
103
        response = self._get_rest_response(rest_name, dpid_id)
104
        json_response = json.loads(response.data)
105
        assert json_response["flow_id"] == flow_id
106
        assert json_response["packet_counter"] == 40
107
        assert json_response["packet_per_second"] == 2.0
108
109
    def test_bytes_count__fail(self):
110
        """Test bytes_count rest call with wrong flow_id."""
111
        flow_id = "123456789"
112
        rest_name = "bytes_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 by
This code seems to be duplicated in your project.
Loading history...
118
    def test_bytes_count(self, mock_from_flow):
119
        """Test bytes_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 = "bytes_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["bytes_counter"] == 10
131
        assert json_response["bits_per_second"] == 4.0
132
133 View Code Duplication
    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
134
    def test_packet_count_per_flow(self, mock_from_flow):
135
        """Test packet_count_per_flow rest call."""
136
        flow_stats = {
137
            'byte_count': 10,
138
            'duration_sec': 20,
139
            'duration_nsec': 30,
140
            'packet_count': 40
141
            }
142
        flow_id = '6055f13593fad45e0b4699f49d56b105'
143
        flow_stats_dict_mock = {flow_id: flow_stats}
144
        dpid_id = "00:00:00:00:00:00:00:01"
145
        flow_by_sw = {dpid_id: flow_stats_dict_mock}
146
        mock_from_flow.return_value = flow_by_sw
147
148
        rest_name = "packet_count/per_flow"
149
        self._patch_switch_flow(flow_id)
150
151
        mock_from_flow.return_value = flow_by_sw
152
        response = self._get_rest_response(rest_name, dpid_id)
153
        json_response = json.loads(response.data)
154
        assert json_response[0]["flow_id"] == flow_id
155
        assert json_response[0]["packet_counter"] == 40
156
        assert json_response[0]["packet_per_second"] == 2.0
157
158 View Code Duplication
    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
159
    def test_bytes_count_per_flow(self, mock_from_flow):
160
        """Test bytes_count_per_flow rest call."""
161
        flow_stats = {
162
            'byte_count': 10,
163
            'duration_sec': 20,
164
            'duration_nsec': 30,
165
            'packet_count': 40
166
            }
167
        flow_id = '6055f13593fad45e0b4699f49d56b105'
168
        flow_stats_dict_mock = {flow_id: flow_stats}
169
        dpid_id = "00:00:00:00:00:00:00:01"
170
        flow_by_sw = {dpid_id: flow_stats_dict_mock}
171
        mock_from_flow.return_value = flow_by_sw
172
173
        rest_name = "bytes_count/per_flow"
174
        self._patch_switch_flow(flow_id)
175
176
        mock_from_flow.return_value = flow_by_sw
177
        response = self._get_rest_response(rest_name, dpid_id)
178
        json_response = json.loads(response.data)
179
        assert json_response[0]["flow_id"] == flow_id
180
        assert json_response[0]["bytes_counter"] == 10
181
        assert json_response[0]["bits_per_second"] == 4.0
182
183 View Code Duplication
    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
184
    def test_flow_stats_by_dpid_flow_id(self, mock_from_flow):
185
        """Test flow_stats rest call."""
186
        flow_stats = {
187
            'byte_count': 148,
188
            'duration_sec': 1589,
189
            'duration_nsec': 556000000,
190
            'packet_count': 2
191
            }
192
        flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_stats}
193
        flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
194
        mock_from_flow.return_value = flow_by_sw
195
196
        api = get_test_client(self.napp.controller, self.napp)
197
        endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
198
        url = f"{self.server_name_url}"+endpoint
199
200
        response = api.get(url)
201
        expected = flow_by_sw
202
        assert response.json == expected
203
        assert response.status_code == 200
204
205 View Code Duplication
    @patch("napps.amlight.flow_stats.main.Main.flow_stats_by_dpid_flow_id")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
206
    def test_flow_stats_by_dpid_flow_id_with_dpid(self, mock_from_flow):
207
        """Test flow_stats rest call."""
208
        flow_stats = {
209
            'byte_count': 148,
210
            'duration_sec': 1589,
211
            'duration_nsec': 556000000,
212
            'packet_count': 2
213
            }
214
        flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_stats}
215
        flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}
216
        mock_from_flow.return_value = flow_by_sw
217
218
        api = get_test_client(self.napp.controller, self.napp)
219
        endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"
220
        url = f"{self.server_name_url}"+endpoint
221
222
        response = api.get(url)
223
        expected = flow_by_sw
224
        assert response.json == expected
225
        assert response.status_code == 200
226
227
    def _patch_switch_flow(self, flow_id):
228
        """Helper method to patch controller to return switch/flow data."""
229
        # patching the flow_stats object in the switch
230
        flow = self._get_mocked_flow_stats()
231
        flow.id = flow_id
232
        switch = MagicMock()
233
        self.napp.controller.switches = {"1": switch}
234
        self.napp.controller.get_switch_by_dpid = MagicMock()
235
        self.napp.controller.get_switch_by_dpid.return_value = switch
236
237
    def _get_rest_response(self, rest_name, url_id):
238
        """Helper method to call a rest endpoint."""
239
        # call rest
240
        api = get_test_client(get_controller_mock(), self.napp)
241
        url = f"{self.server_name_url}/{rest_name}/{url_id}"
242
        response = api.get(url, content_type="application/json")
243
244
        return response
245
246
    def _get_mocked_flow_stats(self):
247
        """Helper method to create a mock flow_stats object."""
248
        flow_stats = MagicMock()
249
        flow_stats.id = 123
250
        flow_stats.byte_count = 10
251
        flow_stats.duration_sec = 20
252
        flow_stats.duration_nsec = 30
253
        flow_stats.packet_count = 40
254
        return flow_stats
255
256
    def _get_mocked_multipart_replies_flows(self):
257
        """Helper method to create mock multipart replies flows"""
258
        flow = self._get_mocked_flow_base()
259
260
        instruction = MagicMock()
261
        flow.instructions = [instruction]
262
263
        replies_flows = [flow]
264
        return replies_flows
265
266
    def _get_mocked_flow_base(self):
267
        """Helper method to create a mock flow object."""
268
        flow = MagicMock()
269
        flow.id = 456
270
        flow.switch = None
271
        flow.table_id = None
272
        flow.match = None
273
        flow.priority = None
274
        flow.idle_timeout = None
275
        flow.hard_timeout = None
276
        flow.cookie = None
277
        flow.stats = self._get_mocked_flow_stats()
278
        return flow
279
280
    @patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
281
    def test_handle_stats_received(self, mock_handle_stats):
282
        """Test handle_stats_received function."""
283
284
        switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
285
        replies_flows = self._get_mocked_multipart_replies_flows()
286
        name = "kytos/of_core.flow_stats.received"
287
        content = {"switch": switch_v0x04, "replies_flows": replies_flows}
288
289
        event = get_kytos_event_mock(name=name, content=content)
290
291
        self.napp.handle_stats_received(event)
292
        mock_handle_stats.assert_called_once()
293
294
    @patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
295
    def test_handle_stats_received_fail(self, mock_handle_stats):
296
        """Test handle_stats_received function for
297
        fail when replies_flows is not in content."""
298
299
        switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
300
        name = "kytos/of_core.flow_stats.received"
301
        content = {"switch": switch_v0x04}
302
303
        event = get_kytos_event_mock(name=name, content=content)
304
305
        self.napp.handle_stats_received(event)
306
        mock_handle_stats.assert_not_called()
307
308
    def test_handle_stats_reply_received(self):
309
        """Test handle_stats_reply_received call."""
310
311
        flows_mock = self._get_mocked_multipart_replies_flows()
312
        self.napp.handle_stats_reply_received(flows_mock)
313
314
        assert list(self.napp.flows_stats_dict.values())[0].id == 456
315