Test Failed
Pull Request — master (#306)
by Vinicius
09:21
created

TestUNI.setUp()   A

Complexity

Conditions 1

Size

Total Lines 7
Code Lines 6

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 1
eloc 6
nop 1
dl 0
loc 7
rs 10
c 0
b 0
f 0
1
"""Interface tests."""
2
import logging
3
import pickle
4
import unittest
5
from unittest.mock import MagicMock, Mock
6
7
from pyof.v0x04.common.port import PortFeatures
8
9
from kytos.core.common import EntityStatus
10
from kytos.core.interface import TAG, UNI, Interface, TAGType
11
from kytos.core.switch import Switch
12
13
logging.basicConfig(level=logging.CRITICAL)
14
15
16
class TestTAG(unittest.TestCase):
17
    """TAG tests."""
18
19
    def setUp(self):
20
        """Create TAG object."""
21
        self.tag = TAG(1, 123)
22
23
    def test_from_dict(self):
24
        """Test from_dict method."""
25
        tag_dict = {'tag_type': 2, 'value': 456}
26
        tag = self.tag.from_dict(tag_dict)
27
28
        self.assertEqual(tag.tag_type, 2)
29
        self.assertEqual(tag.value, 456)
30
31
    def test_as_dict(self):
32
        """Test as_dict method."""
33
        self.assertEqual(self.tag.as_dict(), {'tag_type': 1, 'value': 123})
34
35
    def test_from_json(self):
36
        """Test from_json method."""
37
        tag_json = '{"tag_type": 2, "value": 456}'
38
        tag = self.tag.from_json(tag_json)
39
40
        self.assertEqual(tag.tag_type, 2)
41
        self.assertEqual(tag.value, 456)
42
43
    def test_as_json(self):
44
        """Test as_json method."""
45
        self.assertEqual(self.tag.as_json(), '{"tag_type": 1, "value": 123}')
46
47
    def test__repr__(self):
48
        """Test __repr__ method."""
49
        self.assertEqual(repr(self.tag), 'TAG(<TAGType.VLAN: 1>, 123)')
50
51
52
# pylint: disable=protected-access, too-many-public-methods
53
class TestInterface(unittest.TestCase):
54
    """Test Interfaces."""
55
56
    def setUp(self):
57
        """Create interface object."""
58
        self.iface = self._get_v0x04_iface()
59
60
    def test_repr(self):
61
        """Test repr() output."""
62
        expected = "Interface('name', 42, Switch('dpid'))"
63
        self.assertEqual(repr(self.iface), expected)
64
65
    @staticmethod
66
    def _get_v0x04_iface(*args, **kwargs):
67
        """Create a v0x04 interface object with optional extra arguments."""
68
        switch = Switch('dpid')
69
        switch.connection = Mock()
70
        switch.connection.protocol.version = 0x04
71
        switch.update_lastseen()
72
        return Interface('name', 42, switch, *args, **kwargs)
73
74
    def test_speed_feature_none(self):
75
        """When port's current features is None."""
76
        self.iface.features = None
77
        self.assertIsNone(self.iface.speed)
78
        self.assertEqual('', self.iface.get_hr_speed())
79
80
    def test_speed_feature_zero(self):
81
        """When port's current features is 0. E.g. port 65534."""
82
        self.iface.features = 0
83
        self.assertIsNone(self.iface.speed)
84
        self.assertEqual('', self.iface.get_hr_speed())
85
86
    def test_1_tera_speed(self):
87
        """1Tb link."""
88
        self.iface.features = PortFeatures.OFPPF_1TB_FD
89
        self.assertEqual(10**12 / 8, self.iface.speed)
90
        self.assertEqual('1 Tbps', self.iface.get_hr_speed())
91
92
    def test_100_giga_speed(self):
93
        """100Gb link."""
94
        self.iface.features = PortFeatures.OFPPF_100GB_FD
95
        self.assertEqual(100 * 10**9 / 8, self.iface.speed)
96
        self.assertEqual('100 Gbps', self.iface.get_hr_speed())
97
98
    def test_40_giga_speed(self):
99
        """40Gb link."""
100
        self.iface.features = PortFeatures.OFPPF_40GB_FD
101
        self.assertEqual(40 * 10**9 / 8, self.iface.speed)
102
        self.assertEqual('40 Gbps', self.iface.get_hr_speed())
103
104
    def test_10_giga_speed(self):
105
        """10Gb link."""
106
        self.iface.features = PortFeatures.OFPPF_10GB_FD
107
        self.assertEqual(10 * 10**9 / 8, self.iface.speed)
108
        self.assertEqual('10 Gbps', self.iface.get_hr_speed())
109
110
    def test_1_giga_speed(self):
111
        """1Gb link."""
112
        self.iface.features = PortFeatures.OFPPF_1GB_FD
113
        self.assertEqual(10**9 / 8, self.iface.speed)
114
        self.assertEqual('1 Gbps', self.iface.get_hr_speed())
115
116
    def test_100_mega_speed(self):
117
        """100Mb link."""
118
        self.iface.features = PortFeatures.OFPPF_100MB_FD
119
        self.assertEqual(100 * 10**6 / 8, self.iface.speed)
120
        self.assertEqual('100 Mbps', self.iface.get_hr_speed())
121
122
    def test_10_mega_speed(self):
123
        """10Mb link."""
124
        self.iface.features = PortFeatures.OFPPF_10MB_FD
125
        self.assertEqual(10 * 10**6 / 8, self.iface.speed)
126
        self.assertEqual('10 Mbps', self.iface.get_hr_speed())
127
128
    def test_speed_setter(self):
129
        """Should return speed that was set and not features'."""
130
        expected_speed = 12345
131
        self.iface.set_custom_speed(expected_speed)
132
        actual_speed = self.iface.speed
133
        self.assertEqual(expected_speed, actual_speed)
134
135
    def test_speed_in_constructor(self):
136
        """Custom speed should override features'."""
137
        expected_speed = 6789
138
        iface = self._get_v0x04_iface(speed=expected_speed)
139
        self.assertEqual(expected_speed, iface.speed)
140
141
    def test_speed_removing_features(self):
142
        """Should return custom speed again when features becomes None."""
143
        custom_speed = 101112
144
        of_speed = 10 * 10**6 / 8
145
        iface = self._get_v0x04_iface(speed=custom_speed,
146
                                      features=PortFeatures.OFPPF_10MB_FD)
147
        self.assertEqual(of_speed, iface.speed)
148
        iface.features = None
149
        self.assertEqual(custom_speed, iface.speed)
150
151
    def test_interface_available_tags(self):
152
        """Test available_tags on Interface class."""
153
        default_range = list(range(1, 4096))
154
        intf_values = [tag.value for tag in self.iface.available_tags]
155
        self.assertListEqual(intf_values, default_range)
156
157
        custom_range = list(range(100, 199))
158
        self.iface.set_available_tags(custom_range)
159
        intf_values = [tag.value for tag in self.iface.available_tags]
160
        self.assertListEqual(intf_values, custom_range)
161
162
    def test_all_available_tags(self):
163
        """Test all available_tags on Interface class."""
164
        max_range = 4096
165
166
        for i in range(1, max_range):
167
            next_tag = self.iface.get_next_available_tag()
168
            self.assertIs(type(next_tag), TAG)
169
            self.assertEqual(next_tag.value, max_range - i)
170
171
        next_tag = self.iface.get_next_available_tag()
172
        self.assertEqual(next_tag, False)
173
174
    def test_interface_is_tag_available(self):
175
        """Test is_tag_available on Interface class."""
176
        max_range = 4096
177
        for i in range(1, max_range):
178
            tag = TAG(TAGType.VLAN, i)
179
180
            next_tag = self.iface.is_tag_available(tag)
181
            self.assertTrue(next_tag)
182
183
        # test lower limit
184
        tag = TAG(TAGType.VLAN, 0)
185
        self.assertFalse(self.iface.is_tag_available(tag))
186
        # test upper limit
187
        tag = TAG(TAGType.VLAN, max_range)
188
        self.assertFalse(self.iface.is_tag_available(tag))
189
190
    def test_interface_use_tags(self):
191
        """Test all use_tag on Interface class."""
192
193
        tag = TAG(TAGType.VLAN, 100)
194
        # check use tag for the first time
195
        is_success = self.iface.use_tag(tag)
196
        self.assertTrue(is_success)
197
198
        # check use tag for the second time
199
        is_success = self.iface.use_tag(tag)
200
        self.assertFalse(is_success)
201
202
        # check use tag after returning the tag to the pool
203
        self.iface.make_tag_available(tag)
204
        is_success = self.iface.use_tag(tag)
205
        self.assertTrue(is_success)
206
207
        # Test sanity safe guard
208
        self.assertFalse(self.iface.make_tag_available(None))
209
        self.assertTrue(None not in self.iface.available_tags)
210
211
    def test_enable(self):
212
        """Test enable method."""
213
        self.iface.switch = MagicMock()
214
215
        self.iface.enable()
216
217
        self.iface.switch.enable.assert_called()
218
        self.assertTrue(self.iface._enabled)
219
220
    def test_get_endpoint(self):
221
        """Test get_endpoint method."""
222
        endpoint = ('endpoint', 'time')
223
        self.iface.endpoints = [endpoint]
224
225
        return_endpoint = self.iface.get_endpoint('endpoint')
226
227
        self.assertEqual(return_endpoint, endpoint)
228
229
    def test_add_endpoint(self):
230
        """Test add_endpoint method."""
231
        self.iface.add_endpoint('endpoint')
232
233
        self.assertEqual(len(self.iface.endpoints), 1)
234
235
    def test_delete_endpoint(self):
236
        """Test delete_endpoint method."""
237
        endpoint = ('endpoint', 'time')
238
        self.iface.endpoints = [endpoint]
239
240
        self.iface.delete_endpoint('endpoint')
241
242
        self.assertEqual(len(self.iface.endpoints), 0)
243
244
    def test_update_endpoint(self):
245
        """Test update_endpoint method."""
246
        endpoint = ('endpoint', 'time')
247
        self.iface.endpoints = [endpoint]
248
249
        self.iface.update_endpoint('endpoint')
250
251
        self.assertEqual(len(self.iface.endpoints), 1)
252
253
    def test_update_link__none(self):
254
        """Test update_link method when this interface is not in link
255
           endpoints."""
256
        link = MagicMock()
257
        link.endpoint_a = MagicMock()
258
        link.endpoint_b = MagicMock()
259
260
        result = self.iface.update_link(link)
261
262
        self.assertFalse(result)
263
264
    def test_update_link__endpoint_a(self):
265
        """Test update_link method when this interface is the endpoint a."""
266
        interface = MagicMock()
267
        interface.link = None
268
        link = MagicMock()
269
        link.endpoint_a = self.iface
270
        link.endpoint_b = interface
271
272
        self.iface.update_link(link)
273
274
        self.assertEqual(self.iface.link, link)
275
        self.assertEqual(interface.link, link)
276
277
    def test_update_link__endpoint_b(self):
278
        """Test update_link method when this interface is the endpoint b."""
279
        interface = MagicMock()
280
        interface.link = None
281
        link = MagicMock()
282
        link.endpoint_a = interface
283
        link.endpoint_b = self.iface
284
285
        self.iface.update_link(link)
286
287
        self.assertEqual(self.iface.link, link)
288
        self.assertEqual(interface.link, link)
289
290
    @staticmethod
291
    def test_pickleable_id() -> None:
292
        """Test to make sure the id is pickleable."""
293
        switch = Switch("dpid1")
294
        interface = Interface("s1-eth1", 1, switch)
295
        pickled = pickle.dumps(interface.as_dict())
296
        intf_dict = pickle.loads(pickled)
297
        assert intf_dict["id"] == interface.id
298
299
    def test_status_funcs(self) -> None:
300
        """Test status_funcs."""
301
        self.iface.enable()
302
        self.iface.activate()
303
        assert self.iface.is_active()
304
        Interface.register_status_func("some_napp_some_func",
305
                                       lambda iface: EntityStatus.DOWN)
306
        assert self.iface.status == EntityStatus.DOWN
307
        Interface.register_status_func("some_napp_some_func",
308
                                       lambda iface: None)
309
        assert self.iface.status == EntityStatus.UP
310
311
312
class TestUNI(unittest.TestCase):
313
    """UNI tests."""
314
315
    def setUp(self):
316
        """Create UNI object."""
317
        switch = MagicMock()
318
        switch.id = '00:00:00:00:00:00:00:01'
319
        interface = Interface('name', 1, switch)
320
        user_tag = TAG(1, 123)
321
        self.uni = UNI(interface, user_tag)
322
323
    def test__eq__(self):
324
        """Test __eq__ method."""
325
        user_tag = TAG(2, 456)
326
        interface = Interface('name', 2, MagicMock())
327
        other = UNI(interface, user_tag)
328
329
        self.assertFalse(self.uni == other)
330
331
    def test_is_valid(self):
332
        """Test is_valid method for a valid, invalid and none tag."""
333
        self.assertTrue(self.uni.is_valid())
334
335
        with self.assertRaises(ValueError):
336
            TAG(999999, 123)
337
338
        self.uni.user_tag = None
339
        self.assertTrue(self.uni.is_valid())
340
341
    def test_as_dict(self):
342
        """Test as_dict method."""
343
        expected_dict = {'interface_id': '00:00:00:00:00:00:00:01:1',
344
                         'tag': {'tag_type': 1, 'value': 123}}
345
        self.assertEqual(self.uni.as_dict(), expected_dict)
346
347
    def test_as_json(self):
348
        """Test as_json method."""
349
        expected_json = '{"interface_id": "00:00:00:00:00:00:00:01:1", ' + \
350
                        '"tag": {"tag_type": 1, "value": 123}}'
351
        self.assertEqual(self.uni.as_json(), expected_json)
352