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