1 | """Test kytos.core.switch module.""" |
||
2 | import asyncio |
||
3 | import json |
||
4 | from datetime import datetime |
||
5 | from unittest import TestCase |
||
6 | from unittest.mock import MagicMock, Mock, patch |
||
7 | |||
8 | from kytos.core import Controller |
||
9 | from kytos.core.config import KytosConfig |
||
10 | from kytos.core.constants import FLOOD_TIMEOUT |
||
11 | from kytos.core.interface import Interface |
||
12 | from kytos.core.switch import Switch |
||
13 | |||
14 | |||
15 | def get_date(): |
||
16 | """Return date with FLOOD_TIMEOUT+1 microseconds.""" |
||
17 | return datetime(2000, 1, 1, 0, 0, 0, FLOOD_TIMEOUT+1) |
||
18 | |||
19 | |||
20 | # pylint: disable=protected-access, too-many-public-methods |
||
21 | class TestSwitch(TestCase): |
||
22 | """Switch tests.""" |
||
23 | |||
24 | def setUp(self): |
||
25 | """Instantiate a controller.""" |
||
26 | |||
27 | self.loop = asyncio.new_event_loop() |
||
28 | asyncio.set_event_loop(None) |
||
29 | |||
30 | self.options = KytosConfig().options['daemon'] |
||
31 | self.controller = Controller(self.options, loop=self.loop) |
||
32 | self.controller.log = Mock() |
||
33 | |||
34 | self.switch = self.create_switch() |
||
35 | |||
36 | @staticmethod |
||
37 | def create_switch(): |
||
38 | """Create a new switch.""" |
||
39 | connection = MagicMock() |
||
40 | connection.address = 'addr' |
||
41 | connection.port = 'port' |
||
42 | connection.protocol.version = 0x04 |
||
43 | switch = Switch('00:00:00:00:00:00:00:01', connection) |
||
44 | switch._enabled = True |
||
45 | return switch |
||
46 | |||
47 | def test_repr(self): |
||
48 | """Test repr() output.""" |
||
49 | expected_repr = "Switch('00:00:00:00:00:00:00:01')" |
||
50 | self.assertEqual(repr(self.switch), expected_repr) |
||
51 | |||
52 | def test_id(self): |
||
53 | """Test id property.""" |
||
54 | self.assertEqual(self.switch.id, '00:00:00:00:00:00:00:01') |
||
55 | |||
56 | def test_ofp_version(self): |
||
57 | """Test ofp_version property.""" |
||
58 | self.assertEqual(self.switch.ofp_version, '0x04') |
||
59 | |||
60 | def test_ofp_version__none(self): |
||
61 | """Test ofp_version property when connection is none.""" |
||
62 | self.switch.connection = None |
||
63 | self.assertIsNone(self.switch.ofp_version) |
||
64 | |||
65 | def tearDown(self): |
||
66 | """TearDown.""" |
||
67 | self.loop.close() |
||
68 | |||
69 | def test_switch_vlan_pool_default(self): |
||
70 | """Test default vlan_pool value.""" |
||
71 | self.assertEqual(self.options.vlan_pool, {}) |
||
72 | |||
73 | def test_switch_vlan_pool_options(self): |
||
74 | """Test switch with the example from kytos.conf.""" |
||
75 | dpid = "00:00:00:00:00:00:00:01" |
||
76 | vlan_pool = {"00:00:00:00:00:00:00:01": |
||
77 | {"1": [[1, 2], [5, 10]], "4": [[3, 4]]}} |
||
78 | self.controller.switches[dpid] = self.switch |
||
79 | self.options.vlan_pool = vlan_pool |
||
80 | self.controller.get_switch_or_create(dpid, self.switch.connection) |
||
81 | |||
82 | port_id = 1 |
||
83 | intf = self.controller.switches[dpid].interfaces[port_id] |
||
84 | tag_values = [tag.value for tag in intf.available_tags] |
||
85 | self.assertEqual(tag_values, [1, 5, 6, 7, 8, 9]) |
||
86 | |||
87 | port_id = 4 |
||
88 | intf = self.controller.switches[dpid].interfaces[port_id] |
||
89 | tag_values = [tag.value for tag in intf.available_tags] |
||
90 | self.assertEqual(tag_values, [3]) |
||
91 | |||
92 | # this port number doesn't exist yet. |
||
93 | port_7 = 7 |
||
94 | intf = Interface("test", port_7, self.switch) |
||
95 | # no attr filters, so should associate as it is |
||
96 | self.controller.switches[dpid].update_interface(intf) |
||
97 | intf_obj = self.controller.switches[dpid].interfaces[port_7] |
||
98 | self.assertEqual(intf_obj, intf) |
||
99 | # assert default vlan_pool range (1, 4096) |
||
100 | tag_values = [tag.value for tag in intf_obj.available_tags] |
||
101 | self.assertEqual(tag_values, list(range(1, 4096))) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Loading history...
|
|||
102 | |||
103 | def test_update_description(self): |
||
104 | """Test update_description method.""" |
||
105 | desc = MagicMock() |
||
106 | desc.mfr_desc.value = 'mfr_desc' |
||
107 | desc.hw_desc.value = 'hw_desc' |
||
108 | desc.sw_desc.value = 'sw_desc' |
||
109 | desc.serial_num.value = 'serial_num' |
||
110 | desc.dp_desc.value = 'dp_desc' |
||
111 | |||
112 | self.switch.update_description(desc) |
||
113 | |||
114 | self.assertEqual(self.switch.description['manufacturer'], 'mfr_desc') |
||
115 | self.assertEqual(self.switch.description['hardware'], 'hw_desc') |
||
116 | self.assertEqual(self.switch.description['software'], 'sw_desc') |
||
117 | self.assertEqual(self.switch.description['serial'], 'serial_num') |
||
118 | self.assertEqual(self.switch.description['data_path'], 'dp_desc') |
||
119 | |||
120 | def test_disable(self): |
||
121 | """Test disable method.""" |
||
122 | interface = MagicMock() |
||
123 | self.switch.interfaces = {"1": interface} |
||
124 | |||
125 | self.switch.disable() |
||
126 | |||
127 | interface.disable.assert_called() |
||
128 | self.assertFalse(self.switch._enabled) |
||
129 | |||
130 | def test_disconnect(self): |
||
131 | """Test disconnect method.""" |
||
132 | self.switch.disconnect() |
||
133 | |||
134 | self.assertIsNone(self.switch.connection) |
||
135 | |||
136 | def test_get_interface_by_port_no(self): |
||
137 | """Test get_interface_by_port_no method.""" |
||
138 | interface_1 = MagicMock(port_number='1') |
||
139 | interface_2 = MagicMock(port_number='2') |
||
140 | self.switch.interfaces = {'1': interface_1, '2': interface_2} |
||
141 | |||
142 | expected_interface_1 = self.switch.get_interface_by_port_no('1') |
||
143 | expected_interface_2 = self.switch.get_interface_by_port_no('3') |
||
144 | |||
145 | self.assertEqual(expected_interface_1, interface_1) |
||
146 | self.assertIsNone(expected_interface_2) |
||
147 | |||
148 | def test_get_flow_by_id(self): |
||
149 | """Test get_flow_by_id method.""" |
||
150 | flow_1 = MagicMock(id='1') |
||
151 | flow_2 = MagicMock(id='2') |
||
152 | self.switch.flows = [flow_1, flow_2] |
||
153 | |||
154 | expected_flow_1 = self.switch.get_flow_by_id('1') |
||
155 | expected_flow_2 = self.switch.get_flow_by_id('3') |
||
156 | |||
157 | self.assertEqual(expected_flow_1, flow_1) |
||
158 | self.assertIsNone(expected_flow_2) |
||
159 | |||
160 | def test_is_connected__true(self): |
||
161 | """Test is_connected method.""" |
||
162 | connection = MagicMock() |
||
163 | connection.is_alive.return_value = True |
||
164 | connection.is_established.return_value = True |
||
165 | self.switch.connection = connection |
||
166 | self.switch.is_active = MagicMock() |
||
167 | self.switch.is_active.return_value = True |
||
168 | |||
169 | self.assertTrue(self.switch.is_connected()) |
||
170 | |||
171 | def test_is_connected__not_connection(self): |
||
172 | """Test is_connected method when connection does not exist.""" |
||
173 | self.switch.connection = None |
||
174 | self.switch.is_active = MagicMock() |
||
175 | self.switch.is_active.return_value = True |
||
176 | |||
177 | self.assertFalse(self.switch.is_connected()) |
||
178 | |||
179 | def test_is_connected__not_alive(self): |
||
180 | """Test is_connected method when switch is not active.""" |
||
181 | connection = MagicMock() |
||
182 | connection.is_alive.return_value = True |
||
183 | connection.is_established.return_value = True |
||
184 | self.switch.connection = connection |
||
185 | self.switch.is_active = MagicMock() |
||
186 | self.switch.is_active.return_value = False |
||
187 | |||
188 | self.assertFalse(self.switch.is_connected()) |
||
189 | |||
190 | def test_update_connection(self): |
||
191 | """Test update_connection method.""" |
||
192 | connection = MagicMock() |
||
193 | self.switch.update_connection(connection) |
||
194 | |||
195 | self.assertEqual(self.switch.connection, connection) |
||
196 | self.assertEqual(self.switch.connection.switch, self.switch) |
||
197 | |||
198 | def test_update_features(self): |
||
199 | """Test update_features method.""" |
||
200 | self.switch.update_features('features') |
||
201 | |||
202 | self.assertEqual(self.switch.features, 'features') |
||
203 | |||
204 | def test_send(self): |
||
205 | """Test send method.""" |
||
206 | self.switch.send('buffer') |
||
207 | |||
208 | self.switch.connection.send.assert_called_with('buffer') |
||
209 | |||
210 | @patch('kytos.core.switch.now', return_value=get_date()) |
||
211 | def test_update_lastseen(self, mock_now): |
||
212 | """Test update_lastseen method.""" |
||
213 | self.switch.update_lastseen() |
||
214 | |||
215 | self.assertEqual(self.switch.lastseen, mock_now.return_value) |
||
216 | |||
217 | def test_update_interface(self): |
||
218 | """Test update_interface method.""" |
||
219 | interface = MagicMock(port_number=1) |
||
220 | self.switch.update_interface(interface) |
||
221 | |||
222 | self.assertEqual(self.switch.interfaces[1], interface) |
||
223 | |||
224 | def test_remove_interface(self): |
||
225 | """Test remove_interface method.""" |
||
226 | interface = MagicMock(port_number=1) |
||
227 | self.switch.interfaces[1] = interface |
||
228 | |||
229 | self.switch.remove_interface(interface) |
||
230 | |||
231 | self.assertEqual(self.switch.interfaces, {}) |
||
232 | |||
233 | def test_update_mac_table(self): |
||
234 | """Test update_mac_table method.""" |
||
235 | mac = MagicMock(value='00:00:00:00:00:00') |
||
236 | self.switch.update_mac_table(mac, 1) |
||
237 | self.switch.update_mac_table(mac, 2) |
||
238 | |||
239 | self.assertEqual(self.switch.mac2port[mac.value], {1, 2}) |
||
240 | |||
241 | def test_last_flood(self): |
||
242 | """Test last_flood method.""" |
||
243 | self.switch.flood_table['hash'] = 'timestamp' |
||
244 | ethernet_frame = MagicMock() |
||
245 | ethernet_frame.get_hash.return_value = 'hash' |
||
246 | |||
247 | last_flood = self.switch.last_flood(ethernet_frame) |
||
248 | |||
249 | self.assertEqual(last_flood, 'timestamp') |
||
250 | |||
251 | def test_last_flood__error(self): |
||
252 | """Test last_flood method to error case.""" |
||
253 | ethernet_frame = MagicMock() |
||
254 | ethernet_frame.get_hash.return_value = 'hash' |
||
255 | |||
256 | last_flood = self.switch.last_flood(ethernet_frame) |
||
257 | |||
258 | self.assertIsNone(last_flood) |
||
259 | |||
260 | @patch('kytos.core.switch.now', return_value=get_date()) |
||
261 | def test_should_flood(self, _): |
||
262 | """Test should_flood method.""" |
||
263 | self.switch.flood_table['hash1'] = datetime(2000, 1, 1, 0, 0, 0, 0) |
||
264 | self.switch.flood_table['hash2'] = datetime(2000, 1, 1, 0, 0, 0, |
||
265 | FLOOD_TIMEOUT) |
||
266 | |||
267 | ethernet_frame = MagicMock() |
||
268 | ethernet_frame.get_hash.side_effect = ['hash1', 'hash2'] |
||
269 | |||
270 | should_flood_1 = self.switch.should_flood(ethernet_frame) |
||
271 | should_flood_2 = self.switch.should_flood(ethernet_frame) |
||
272 | |||
273 | self.assertTrue(should_flood_1) |
||
274 | self.assertFalse(should_flood_2) |
||
275 | |||
276 | @patch('kytos.core.switch.now', return_value=get_date()) |
||
277 | def test_update_flood_table(self, mock_now): |
||
278 | """Test update_flood_table method.""" |
||
279 | ethernet_frame = MagicMock() |
||
280 | ethernet_frame.get_hash.return_value = 'hash' |
||
281 | |||
282 | self.switch.update_flood_table(ethernet_frame) |
||
283 | |||
284 | self.assertEqual(self.switch.flood_table['hash'], |
||
285 | mock_now.return_value) |
||
286 | |||
287 | def test_where_is_mac(self): |
||
288 | """Test where_is_mac method.""" |
||
289 | mac = MagicMock(value='00:00:00:00:00:00') |
||
290 | |||
291 | expected_ports_1 = self.switch.where_is_mac(mac) |
||
292 | |||
293 | self.switch.mac2port['00:00:00:00:00:00'] = set([1, 2, 3]) |
||
294 | expected_ports_2 = self.switch.where_is_mac(mac) |
||
295 | |||
296 | self.assertIsNone(expected_ports_1) |
||
297 | self.assertEqual(expected_ports_2, [1, 2, 3]) |
||
298 | |||
299 | def test_as_dict(self): |
||
300 | """Test as_dict method.""" |
||
301 | expected_dict = {'id': '00:00:00:00:00:00:00:01', |
||
302 | 'name': '00:00:00:00:00:00:00:01', |
||
303 | 'dpid': '00:00:00:00:00:00:00:01', |
||
304 | 'connection': 'addr:port', |
||
305 | 'ofp_version': '0x04', |
||
306 | 'type': 'switch', |
||
307 | 'manufacturer': '', |
||
308 | 'serial': '', |
||
309 | 'hardware': '', |
||
310 | 'software': None, |
||
311 | 'data_path': '', |
||
312 | 'interfaces': {}, |
||
313 | 'metadata': {}, |
||
314 | 'active': True, |
||
315 | 'enabled': True} |
||
316 | self.assertEqual(self.switch.as_dict(), expected_dict) |
||
317 | |||
318 | def test_as_json(self): |
||
319 | """Test as_json method.""" |
||
320 | expected_json = json.dumps({'id': '00:00:00:00:00:00:00:01', |
||
321 | 'name': '00:00:00:00:00:00:00:01', |
||
322 | 'dpid': '00:00:00:00:00:00:00:01', |
||
323 | 'connection': 'addr:port', |
||
324 | 'ofp_version': '0x04', |
||
325 | 'type': 'switch', |
||
326 | 'manufacturer': '', |
||
327 | 'serial': '', |
||
328 | 'hardware': '', |
||
329 | 'software': None, |
||
330 | 'data_path': '', |
||
331 | 'interfaces': {}, |
||
332 | 'metadata': {}, |
||
333 | 'active': True, |
||
334 | 'enabled': True}) |
||
335 | |||
336 | self.assertEqual(self.switch.as_json(), expected_json) |
||
337 |