Passed
Push — master ( 9ccb0e...0c149d )
by Humberto
05:05 queued 02:22
created

TestInterface.test_speed_removing_features()   A

Complexity

Conditions 1

Size

Total Lines 9
Code Lines 8

Duplication

Lines 0
Ratio 0 %

Importance

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