Passed
Push — master ( c58985...e06700 )
by
unknown
01:57 queued 12s
created

TestUtils.test_compare_endpoints1()   B

Complexity

Conditions 1

Size

Total Lines 163
Code Lines 97

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 3
CRAP Score 1

Importance

Changes 0
Metric Value
eloc 97
dl 0
loc 163
ccs 3
cts 3
cp 1
rs 7.0981
c 0
b 0
f 0
cc 1
nop 4
crap 1

How to fix   Long Method   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

1
"""Module to test the utils.py file."""
2 1
import pytest
3 1
from unittest.mock import patch, MagicMock
4
5 1
from kytos.core.interface import Interface
6 1
from kytos.lib.helpers import get_link_mock
7 1
from napps.amlight.sdntrace_cp import utils, settings
8
9
10
# pylint: disable=too-many-public-methods, duplicate-code, protected-access
11 1
class TestUtils():
12
    """Test utils.py functions."""
13
14 1
    @patch("requests.get")
15 1
    def test_get_stored_flows(self, get_mock):
16
        "Test get_stored_flows"
17 1
        response = MagicMock()
18 1
        response.status_code = 200
19 1
        response.json.return_value = {"result": "ok"}
20 1
        get_mock.return_value = response
21
22 1
        api_url = f'{settings.FLOW_MANAGER_URL}/stored_flows/?state=installed'
23 1
        result = utils.get_stored_flows()
24 1
        get_mock.assert_called_with(api_url, timeout=20)
25 1
        assert result['result'] == "ok"
26
27 1
    def test_convert_list_entries(self):
28
        """Verify convert entries with a list of one example"""
29 1
        eth = {"dl_vlan": 100}
30 1
        dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
31 1
        switch = {"switch": dpid, "eth": eth}
32 1
        entries = {"trace": switch}
33
34 1
        result = utils.convert_list_entries([entries])
35 1
        expected = [{
36
                "dpid": "00:00:00:00:00:00:00:01",
37
                "in_port": 1,
38
                "dl_vlan": [100],
39
            }]
40 1
        assert result == expected
41
42 1
    def test_convert_entries_vlan(self):
43
        """Verify convert entries with simple example with vlan."""
44
45 1
        eth = {"dl_vlan": 100}
46 1
        dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
47 1
        switch = {"switch": dpid, "eth": eth}
48 1
        entries = {"trace": switch}
49
50 1
        result = utils.convert_entries(entries)
51 1
        expected = {
52
                "dpid": "00:00:00:00:00:00:00:01",
53
                "in_port": 1,
54
                "dl_vlan": [100],
55
            }
56 1
        assert result == expected
57
58 1
    def test_prepare_json(self):
59
        """Verify prepare json with simple tracepath result."""
60 1
        trace_result = [
61
            {
62
                "in": {
63
                    "dpid": "00:00:00:00:00:00:00:01",
64
                    "port": 1,
65
                    "time": "2022-06-01 01:01:01.100000",
66
                    "type": "starting",
67
                }
68
            },
69
            {
70
                "in": {
71
                    "dpid": "00:00:00:00:00:00:00:03",
72
                    "port": 3,
73
                    "time": "2022-06-01 01:01:01.100000",
74
                    "type": "intermediary",
75
                    "vlan": 100,
76
                },
77
                "out": {
78
                    "port": 1,
79
                    "vlan": 123,
80
                },
81
            }
82
        ]
83
84 1
        result = utils.prepare_json(trace_result)
85 1
        expected = {
86
                "result": [
87
                    {
88
                        "dpid": "00:00:00:00:00:00:00:01",
89
                        "port": 1,
90
                        "time": "2022-06-01 01:01:01.100000",
91
                        "type": "starting",
92
                    },
93
                    {
94
                        "dpid": "00:00:00:00:00:00:00:03",
95
                        "port": 3,
96
                        "time": "2022-06-01 01:01:01.100000",
97
                        "type": "intermediary",
98
                        "vlan": 100,
99
                        "out": {"port": 1, "vlan": 123},
100
                    },
101
                ]
102
            }
103 1
        assert result == expected
104
105 1
    def test_prepare_list_json(self):
106
        """Verify prepare list with a simple tracepath result."""
107 1
        trace_result = [
108
            {
109
                "in": {
110
                    "dpid": "00:00:00:00:00:00:00:01",
111
                    "port": 1,
112
                    "time": "2022-06-01 01:01:01.100000",
113
                    "type": "starting",
114
                }
115
            },
116
            {
117
                "in": {
118
                    "dpid": "00:00:00:00:00:00:00:03",
119
                    "port": 3,
120
                    "time": "2022-06-01 01:01:01.100000",
121
                    "type": "intermediary",
122
                    "vlan": 100,
123
                },
124
                "out": {
125
                    "port": 1,
126
                    "vlan": 123,
127
                },
128
            }
129
        ]
130
131 1
        result = utils._prepare_json(trace_result)
132 1
        expected = [
133
                    {
134
                        "dpid": "00:00:00:00:00:00:00:01",
135
                        "port": 1,
136
                        "time": "2022-06-01 01:01:01.100000",
137
                        "type": "starting",
138
                    },
139
                    {
140
                        "dpid": "00:00:00:00:00:00:00:03",
141
                        "port": 3,
142
                        "time": "2022-06-01 01:01:01.100000",
143
                        "type": "intermediary",
144
                        "vlan": 100,
145
                        "out": {"port": 1, "vlan": 123},
146
                    },
147
                ]
148 1
        assert result == expected
149
150 1
    def test_prepare_json_empty(self):
151
        """Verify prepare json with empty result."""
152 1
        trace_result = []
153
154 1
        result = utils.prepare_json(trace_result)
155
156 1
        assert result == {"result": []}
157
158 1
    @pytest.mark.parametrize(
159
        "endpoint1,endpoint2,result",
160
        [
161
            (
162
                {"dpid": "00:00:00:00:00:00:00:01"},
163
                {"dpid": "00:00:00:00:00:00:00:02"},
164
                False
165
            ),
166
            (
167
                {
168
                    "dpid": "00:00:00:00:00:00:00:03",
169
                    "out_port": 2,
170
                    "out_vlan": 200,
171
                },
172
                {
173
                    "dpid": "00:00:00:00:00:00:00:03",
174
                    "in_port": 3,
175
                    "in_vlan": 100,
176
                },
177
                False
178
            ),
179
            (
180
                {
181
                    "dpid": "00:00:00:00:00:00:00:03",
182
                    "in_port": 3,
183
                    "in_vlan": 100,
184
                },
185
                {
186
                    "dpid": "00:00:00:00:00:00:00:03",
187
                    "in_port": 3,
188
                    "in_vlan": 100,
189
                },
190
                False
191
            ),
192
            (
193
                {
194
                    "dpid": "00:00:00:00:00:00:00:03",
195
                    "in_port": 3,
196
                    "in_vlan": 100,
197
                },
198
                {
199
                    "dpid": "00:00:00:00:00:00:00:03",
200
                    "out_port": 2,
201
                    "out_vlan": 200,
202
                },
203
                False
204
            ),
205
            (
206
                {
207
                    "dpid": "00:00:00:00:00:00:00:03",
208
                    "in_port": 3,
209
                    "out_port": 2,
210
                    "in_vlan": 100,
211
                },
212
                {
213
                    "dpid": "00:00:00:00:00:00:00:03",
214
                    "in_port": 2,
215
                    "out_port": 3,
216
                    "out_vlan": 200,
217
                },
218
                False
219
            ),
220
            (
221
                {
222
                    "dpid": "00:00:00:00:00:00:00:03",
223
                    "in_port": 3,
224
                    "out_port": 2,
225
                    "in_vlan": 100,
226
                },
227
                {
228
                    "dpid": "00:00:00:00:00:00:00:03",
229
                    "in_port": 2,
230
                    "out_port": 3,
231
                },
232
                False
233
            ),
234
            (
235
                {
236
                    "dpid": "00:00:00:00:00:00:00:03",
237
                    "in_port": 3,
238
                    "out_port": 2,
239
                },
240
                {
241
                    "dpid": "00:00:00:00:00:00:00:03",
242
                    "in_port": 2,
243
                    "out_port": 3,
244
                    "out_vlan": 200,
245
                },
246
                False
247
            ),
248
            (
249
                {
250
                    "dpid": "00:00:00:00:00:00:00:01",
251
                    "in_port": 3,
252
                    "out_port": 2,
253
                    "out_vlan": 200,
254
                },
255
                {
256
                    "dpid": "00:00:00:00:00:00:00:01",
257
                    "in_port": 2,
258
                    "out_port": 3,
259
                    "in_vlan": 100,
260
                },
261
                False
262
            ),
263
            (
264
                {
265
                    "dpid": "00:00:00:00:00:00:00:01",
266
                    "in_port": 3,
267
                    "out_port": 2,
268
                    "out_vlan": 200,
269
                },
270
                {
271
                    "dpid": "00:00:00:00:00:00:00:01",
272
                    "in_port": 2,
273
                    "out_port": 3,
274
                },
275
                False
276
            ),
277
            (
278
                {
279
                    "dpid": "00:00:00:00:00:00:00:01",
280
                    "in_port": 3,
281
                    "out_port": 2,
282
                },
283
                {
284
                    "dpid": "00:00:00:00:00:00:00:01",
285
                    "in_port": 2,
286
                    "out_port": 3,
287
                    "in_vlan": 100,
288
                },
289
                False
290
            ),
291
            (
292
                {
293
                    "dpid": "00:00:00:00:00:00:00:03",
294
                    "in_port": 3,
295
                    "in_vlan": 100,
296
                },
297
                {
298
                    "dpid": "00:00:00:00:00:00:00:03",
299
                    "out_port": 3,
300
                    "out_vlan": 100,
301
                },
302
                True
303
            ),
304
        ]
305
    )
306 1
    def test_compare_endpoints1(self, endpoint1, endpoint2, result):
307
        """Test for compare endpoinst for the internal conditional no.
308
        1 - first: Test endpoint1 dpid != endpoint2 dpid
309
        2 - second: Test endpoint1 without in_port
310
        3 - second: Test endpoint2 without out_port
311
        4 - second: Test endpoint1 in_port != endpoint2 out_port
312
        5 - third: Test endpoint1 in_vlan != endpoint2 out_vlan
313
        6 - first: Test endpoint1 with in_vlan and endpoint2 without out_vlan
314
        7 - first: Test endpoint1 without in_vlan and endpoint2 with out_vlan
315
        8 - fifth: Test endpoint1 out_vlan != endpoint2 in_vlan
316
        9 - fifth: Test endpoint1 with out_vlan and endpoint2 without in_vlan
317
        10 - fifth: Test endpoint1 without out_vlan and endpoint2 with in_vlan
318
        11 - fifth: Test endpoint1 out_vlan != endpoint2 in_vlan
319
        """
320 1
        assert utils._compare_endpoints(endpoint1, endpoint2) == result
321
322 1
    def test_find_endpoint_b(self):
323
        """Test find endpoint with interface equals link endpoint B."""
324 1
        port = 1
325
326 1
        mock_interface = Interface("interface A", port, MagicMock())
327 1
        mock_interface.address = "00:00:00:00:00:00:00:01"
328 1
        mock_interface.link = get_link_mock(
329
            "00:00:00:00:00:00:00:02", "00:00:00:00:00:00:00:01"
330
        )
331
332 1
        mock_switch = MagicMock()
333 1
        mock_switch.get_interface_by_port_no.return_value = mock_interface
334 1
        expected = {'endpoint': mock_interface.link.endpoint_a}
335 1
        result = utils.find_endpoint(mock_switch, port)
336 1
        assert result == expected
337
338 1
    def test_find_endpoint_a(self):
339
        """Test find endpoint with interface equals link endpoint A."""
340 1
        port = 1
341
342 1
        mock_interface = Interface("interface A", port, MagicMock())
343 1
        mock_interface.address = "00:00:00:00:00:00:00:01"
344 1
        mock_interface.link = get_link_mock(
345
            "00:00:00:00:00:00:00:01", "00:00:00:00:00:00:00:03"
346
        )
347
348 1
        mock_switch = MagicMock()
349 1
        mock_switch.get_interface_by_port_no.return_value = mock_interface
350 1
        expected = {'endpoint': mock_interface.link.endpoint_b}
351 1
        result = utils.find_endpoint(mock_switch, port)
352 1
        assert result == expected
353
354 1
    def test_find_endpoint_link_none(self):
355
        """Test find endpoint without link."""
356 1
        port = 1
357
358 1
        mock_interface = Interface("interface A", port, MagicMock())
359 1
        mock_interface.address = "00:00:00:00:00:00:00:01"
360
361 1
        mock_switch = MagicMock()
362 1
        mock_switch.get_interface_by_port_no.return_value = mock_interface
363
364 1
        result = utils.find_endpoint(mock_switch, port)
365 1
        assert 'endpoint' in result
366 1
        assert result['endpoint'] is None
367
368 1
    def test_convert_vlan(self):
369
        """Test convert_vlan function"""
370 1
        value = 100
371 1
        result = utils.convert_vlan(value)
372 1
        assert result[0] == 100
373
374 1
        value = "4096/4096"
375 1
        result = utils.convert_vlan(value)
376 1
        assert result[0] == 4096
377 1
        assert result[1] == 4096
378
379 1
    @pytest.mark.parametrize(
380
        "value,field_flow,result",
381
        [
382
            (None, 0, True),
383
            (None, 10, False),
384
            (None, "4096/4096", False),
385
            ([10], 0, False),
386
            ([10], 10, True),
387
            ([10], "4096/4096", True),
388
            ([10], 11, False),
389
            ([3], "5/1", True),
390
            ([2], "5/1", False),
391
        ]
392
    )
393 1
    def test_match_field_dl_vlan(self, value, field_flow, result):
394
        """Test match_field_dl_vlan"""
395 1
        assert utils.match_field_dl_vlan(value, field_flow) == result
396
397 1
    @pytest.mark.parametrize(
398
        "field,field_flow,result",
399
        [
400
            ('192.168.20.21', '192.168.20.21', True),
401
            ('192.168.20.21', '192.168.20.21/10', True),
402
            ('192.168.20.21', '192.168.20.21/32', True),
403
            ('192.168.20.21', '192.168.20.21/255.255.255.255', True),
404
            ('192.168.20.30', '192.168.20.21', False),
405
            ('192.200.20.30', '192.168.20.21/10', False),
406
            ('2002:db8::8a3f:362:7897', '2002:db8::8a3f:362:7897', True),
407
            ('2002:db8::8a3f:362:7897', '2002:db8::8a3f:362:7897/10', True),
408
            ('2002:db8::8a3f:362:7897', '2002:db8::8a3f:362:7897/128', True),
409
            ('2002:db8::8a3f:362:7', '2002:db8::8a3f:362:7897', False),
410
            ('3002:db8::9a3f:362:7897', '2002:db8::8a3f:362:7897/10', False),
411
        ],
412
    )
413 1
    def test_match_field_ip(self, field, field_flow, result):
414
        """Test match_field_ip"""
415
        assert utils.match_field_ip(field, field_flow) == result
416