1 # Copyright (c) 2018 Intel Corporation
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
19 from yardstick.common import exceptions
20 from yardstick.network_services.libs.ixia_libs.ixnet import ixnet_api
26 TRAFFIC_PARAMETERS = {
34 'framesize': {'64B': '25', '256B': '75'}
39 'dstip4': '152.16.40.20',
41 'srcip4': '152.16.100.20',
46 'dstip4': '152.16.40.20',
48 'srcip4': '152.16.100.20',
54 'dstip4': '152.16.100.20',
56 'srcip4': '152.16.40.20',
63 'traffic_type': 'continuous'
72 'framesize': {'128B': '35', '1024B': '65'}
77 'dstip4': '152.16.100.20',
79 'srcip4': '152.16.40.20',
85 'dstip4': '152.16.100.20',
87 'srcip4': '152.16.40.20',
93 'dstip4': '152.16.100.20',
95 'srcip4': '152.16.40.20',
102 'traffic_type': 'continuous'
107 class TestIxNextgen(unittest.TestCase):
110 self.ixnet = mock.Mock()
111 self.ixnet.execute = mock.Mock()
112 self.ixnet.getRoot.return_value = 'my_root'
114 def test_get_config(self):
118 'external-interface': [
119 {'virtual-interface': {'vpci': '0000:07:00.1'}},
120 {'virtual-interface': {'vpci': '0001:08:01.2'}}
127 'dut_result_dir': 'test2',
129 'ixchassis': 'test4',
139 'cards': ['0000', '0001'],
140 'ports': ['07', '08'],
141 'output_dir': 'test2',
146 result = ixnet_api.IxNextgen.get_config(tg_cfg)
147 self.assertEqual(result, expected)
149 def test__get_config_element_by_flow_group_name(self):
150 ixnet_gen = ixnet_api.IxNextgen()
151 ixnet_gen._ixnet = self.ixnet
152 ixnet_gen._ixnet.getList.side_effect = [['traffic_item'],
154 ixnet_gen._ixnet.getAttribute.return_value = 'flow_group_01'
155 output = ixnet_gen._get_config_element_by_flow_group_name(
157 self.assertEqual('traffic_item/configElement:flow_group_01', output)
159 def test__get_config_element_by_flow_group_name_no_match(self):
160 ixnet_gen = ixnet_api.IxNextgen()
161 ixnet_gen._ixnet = self.ixnet
162 ixnet_gen._ixnet.getList.side_effect = [['traffic_item'],
164 ixnet_gen._ixnet.getAttribute.return_value = 'flow_group_02'
165 output = ixnet_gen._get_config_element_by_flow_group_name(
167 self.assertIsNone(output)
169 def test__get_stack_item(self):
170 ixnet_gen = ixnet_api.IxNextgen()
171 ixnet_gen._ixnet = self.ixnet
172 ixnet_gen._ixnet.getList.return_value = ['tcp1', 'tcp2', 'udp']
173 with mock.patch.object(
174 ixnet_gen, '_get_config_element_by_flow_group_name') as \
175 mock_get_cfg_element:
176 mock_get_cfg_element.return_value = 'cfg_element'
177 output = ixnet_gen._get_stack_item(mock.ANY, ixnet_api.PROTO_TCP)
178 self.assertEqual(['tcp1', 'tcp2'], output)
180 def test__get_stack_item_no_config_element(self):
181 ixnet_gen = ixnet_api.IxNextgen()
182 ixnet_gen._ixnet = self.ixnet
183 with mock.patch.object(
184 ixnet_gen, '_get_config_element_by_flow_group_name',
186 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
187 ixnet_gen._get_stack_item(mock.ANY, mock.ANY)
189 def test__get_field_in_stack_item(self):
190 ixnet_gen = ixnet_api.IxNextgen()
191 ixnet_gen._ixnet = self.ixnet
192 ixnet_gen._ixnet.getList.return_value = ['field1', 'field2']
193 output = ixnet_gen._get_field_in_stack_item(mock.ANY, 'field2')
194 self.assertEqual('field2', output)
196 def test__get_field_in_stack_item_no_field_present(self):
197 ixnet_gen = ixnet_api.IxNextgen()
198 ixnet_gen._ixnet = self.ixnet
199 ixnet_gen._ixnet.getList.return_value = ['field1', 'field2']
200 with self.assertRaises(exceptions.IxNetworkFieldNotPresentInStackItem):
201 ixnet_gen._get_field_in_stack_item(mock.ANY, 'field3')
203 def test__parse_framesize(self):
204 ixnet_gen = ixnet_api.IxNextgen()
205 ixnet_gen._ixnet = self.ixnet
206 framesize = {'64B': '75', '512b': '25'}
207 output = ixnet_gen._parse_framesize(framesize)
208 self.assertEqual(2, len(output))
209 self.assertIn([64, 64, 75], output)
210 self.assertIn([512, 512, 25], output)
212 @mock.patch.object(IxNetwork, 'IxNet')
213 def test_connect(self, mock_ixnet):
214 mock_ixnet.return_value = self.ixnet
215 ixnet_gen = ixnet_api.IxNextgen()
216 with mock.patch.object(ixnet_gen, 'get_config') as mock_config:
217 mock_config.return_value = {'machine': 'machine_fake',
220 ixnet_gen.connect(mock.ANY)
222 self.ixnet.connect.assert_called_once_with(
223 'machine_fake', '-port', 'port_fake', '-version', '12345')
224 mock_config.assert_called_once()
226 def test_connect_invalid_config_no_machine(self):
227 ixnet_gen = ixnet_api.IxNextgen()
228 ixnet_gen._ixnet = self.ixnet
229 ixnet_gen.get_config = mock.Mock(return_value={
232 self.assertRaises(KeyError, ixnet_gen.connect, mock.ANY)
233 self.ixnet.connect.assert_not_called()
235 def test_connect_invalid_config_no_port(self):
236 ixnet_gen = ixnet_api.IxNextgen()
237 ixnet_gen._ixnet = self.ixnet
238 ixnet_gen.get_config = mock.Mock(return_value={
239 'machine': 'machine_fake',
241 self.assertRaises(KeyError, ixnet_gen.connect, mock.ANY)
242 self.ixnet.connect.assert_not_called()
244 def test_connect_invalid_config_no_version(self):
245 ixnet_gen = ixnet_api.IxNextgen()
246 ixnet_gen._ixnet = self.ixnet
247 ixnet_gen.get_config = mock.Mock(return_value={
248 'machine': 'machine_fake',
249 'port': 'port_fake'})
250 self.assertRaises(KeyError, ixnet_gen.connect, mock.ANY)
251 self.ixnet.connect.assert_not_called()
253 def test_connect_no_config(self):
254 ixnet_gen = ixnet_api.IxNextgen()
255 ixnet_gen._ixnet = self.ixnet
256 ixnet_gen.get_config = mock.Mock(return_value={})
257 self.assertRaises(KeyError, ixnet_gen.connect, mock.ANY)
258 self.ixnet.connect.assert_not_called()
260 def test_clear_config(self):
261 ixnet_gen = ixnet_api.IxNextgen()
262 ixnet_gen._ixnet = self.ixnet
263 ixnet_gen.clear_config()
264 self.ixnet.execute.assert_called_once_with('newConfig')
266 @mock.patch.object(ixnet_api, 'log')
267 def test_assign_ports_2_ports(self, *args):
268 self.ixnet.getAttribute.side_effect = ['up', 'down']
270 'chassis': '1.1.1.1',
273 ixnet_gen = ixnet_api.IxNextgen()
274 ixnet_gen._ixnet = self.ixnet
275 ixnet_gen._cfg = config
277 self.assertIsNone(ixnet_gen.assign_ports())
278 self.assertEqual(self.ixnet.execute.call_count, 2)
279 self.assertEqual(self.ixnet.commit.call_count, 4)
280 self.assertEqual(self.ixnet.getAttribute.call_count, 2)
282 @mock.patch.object(ixnet_api, 'log')
283 def test_assign_ports_port_down(self, mock_log):
284 self.ixnet.getAttribute.return_value = 'down'
286 'chassis': '1.1.1.1',
289 ixnet_gen = ixnet_api.IxNextgen()
290 ixnet_gen._ixnet = self.ixnet
291 ixnet_gen._cfg = config
292 ixnet_gen.assign_ports()
293 mock_log.warning.assert_called()
295 def test_assign_ports_no_config(self):
296 ixnet_gen = ixnet_api.IxNextgen()
297 ixnet_gen._ixnet = self.ixnet
299 self.assertRaises(KeyError, ixnet_gen.assign_ports)
301 def test__create_traffic_item(self):
302 ixnet_gen = ixnet_api.IxNextgen()
303 ixnet_gen._ixnet = self.ixnet
304 self.ixnet.add.return_value = 'my_new_traffic_item'
305 self.ixnet.remapIds.return_value = ['my_traffic_item_id']
307 ixnet_gen._create_traffic_item()
308 self.ixnet.add.assert_called_once_with(
309 'my_root/traffic', 'trafficItem')
310 self.ixnet.setMultiAttribute.assert_called_once_with(
311 'my_new_traffic_item', '-name', 'RFC2544', '-trafficType', 'raw')
312 self.assertEqual(2, self.ixnet.commit.call_count)
313 self.ixnet.remapIds.assert_called_once_with('my_new_traffic_item')
314 self.ixnet.setAttribute('my_traffic_item_id/tracking',
315 '-trackBy', 'trafficGroupId0')
317 def test__create_flow_groups(self):
318 ixnet_gen = ixnet_api.IxNextgen()
319 ixnet_gen._ixnet = self.ixnet
320 ixnet_gen.ixnet.getList.side_effect = [['traffic_item'], ['1', '2']]
321 ixnet_gen.ixnet.add.side_effect = ['endp1', 'endp2']
322 ixnet_gen._create_flow_groups()
323 ixnet_gen.ixnet.add.assert_has_calls([
324 mock.call('traffic_item', 'endpointSet'),
325 mock.call('traffic_item', 'endpointSet')])
326 ixnet_gen.ixnet.setMultiAttribute.assert_has_calls([
327 mock.call('endp1', '-name', '1', '-sources', ['1/protocols'],
328 '-destinations', ['2/protocols']),
329 mock.call('endp2', '-name', '2', '-sources', ['2/protocols'],
330 '-destinations', ['1/protocols'])])
332 def test__append_protocol_to_stack(self):
333 ixnet_gen = ixnet_api.IxNextgen()
334 ixnet_gen._ixnet = self.ixnet
336 ixnet_gen._append_procotol_to_stack('my_protocol', 'prev_element')
337 self.ixnet.execute.assert_called_with(
338 'append', 'prev_element',
339 'my_root/traffic/protocolTemplate:"my_protocol"')
341 def test__setup_config_elements(self):
342 ixnet_gen = ixnet_api.IxNextgen()
343 ixnet_gen._ixnet = self.ixnet
344 ixnet_gen.ixnet.getList.side_effect = [['traffic_item'],
346 with mock.patch.object(ixnet_gen, '_append_procotol_to_stack') as \
348 ixnet_gen._setup_config_elements()
349 mock_append_proto.assert_has_calls([
350 mock.call(ixnet_api.PROTO_UDP, 'cfg_element/stack:"ethernet-1"'),
351 mock.call(ixnet_api.PROTO_IPV4, 'cfg_element/stack:"ethernet-1"')])
352 ixnet_gen.ixnet.setAttribute.assert_has_calls([
353 mock.call('cfg_element/frameRateDistribution', '-portDistribution',
355 mock.call('cfg_element/frameRateDistribution',
356 '-streamDistribution', 'splitRateEvenly')])
358 @mock.patch.object(ixnet_api.IxNextgen, '_create_traffic_item')
359 @mock.patch.object(ixnet_api.IxNextgen, '_create_flow_groups')
360 @mock.patch.object(ixnet_api.IxNextgen, '_setup_config_elements')
361 def test_create_traffic_model(self, mock__setup_config_elements,
362 mock__create_flow_groups,
363 mock__create_traffic_item):
364 ixnet_gen = ixnet_api.IxNextgen()
365 ixnet_gen._ixnet = self.ixnet
367 ixnet_gen.create_traffic_model()
368 mock__create_traffic_item.assert_called_once()
369 mock__create_flow_groups.assert_called_once()
370 mock__setup_config_elements.assert_called_once()
372 def test__update_frame_mac(self):
373 ixnet_gen = ixnet_api.IxNextgen()
374 ixnet_gen._ixnet = self.ixnet
375 with mock.patch.object(ixnet_gen, '_get_field_in_stack_item') as \
377 mock_get_field.return_value = 'field_descriptor'
378 ixnet_gen._update_frame_mac('ethernet_descriptor', 'field', 'mac')
379 mock_get_field.assert_called_once_with('ethernet_descriptor', 'field')
380 ixnet_gen.ixnet.setMultiAttribute(
381 'field_descriptor', '-singleValue', 'mac', '-fieldValue', 'mac',
382 '-valueType', 'singleValue')
383 ixnet_gen.ixnet.commit.assert_called_once()
385 def test_update_frame(self):
386 ixnet_gen = ixnet_api.IxNextgen()
387 ixnet_gen._ixnet = self.ixnet
388 with mock.patch.object(
389 ixnet_gen, '_get_config_element_by_flow_group_name',
390 return_value='cfg_element'), \
391 mock.patch.object(ixnet_gen, '_update_frame_mac') as \
393 mock.patch.object(ixnet_gen, '_get_stack_item') as \
395 mock_get_stack_item.side_effect = [['item1'], ['item2'],
396 ['item3'], ['item4']]
397 ixnet_gen.update_frame(TRAFFIC_PARAMETERS)
399 self.assertEqual(6, len(ixnet_gen.ixnet.setMultiAttribute.mock_calls))
400 self.assertEqual(4, len(mock_update_frame.mock_calls))
401 ixnet_gen.ixnet.setMultiAttribute.assert_has_calls(
402 [mock.call('cfg_element/frameRate', '-rate', 10000.5,
403 '-type', 'framesPerSecond'),
404 mock.call('cfg_element/frameRate', '-rate', 75.2, '-type',
408 def test_update_frame_flow_not_present(self):
409 ixnet_gen = ixnet_api.IxNextgen()
410 ixnet_gen._ixnet = self.ixnet
411 with mock.patch.object(
412 ixnet_gen, '_get_config_element_by_flow_group_name',
414 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
415 ixnet_gen.update_frame(TRAFFIC_PARAMETERS)
417 def test_get_statistics(self):
418 ixnet_gen = ixnet_api.IxNextgen()
419 port_statistics = '::ixNet::OBJ-/statistics/view:"Port Statistics"'
420 flow_statistics = '::ixNet::OBJ-/statistics/view:"Flow Statistics"'
421 with mock.patch.object(ixnet_gen, '_build_stats_map') as \
423 ixnet_gen.get_statistics()
425 mock_build_stats.assert_has_calls([
426 mock.call(port_statistics, ixnet_gen.PORT_STATS_NAME_MAP),
427 mock.call(flow_statistics, ixnet_gen.LATENCY_NAME_MAP)])
429 def test__update_ipv4_address(self):
430 ixnet_gen = ixnet_api.IxNextgen()
431 ixnet_gen._ixnet = self.ixnet
432 with mock.patch.object(ixnet_gen, '_get_field_in_stack_item',
433 return_value='field_desc'):
434 ixnet_gen._update_ipv4_address(mock.ANY, mock.ANY, '192.168.1.1',
435 100, '255.255.255.0', 25)
436 ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with(
437 'field_desc', '-seed', 100, '-fixedBits', '192.168.1.1',
438 '-randomMask', '255.255.255.0', '-valueType', 'random',
441 def test_update_ip_packet(self):
442 ixnet_gen = ixnet_api.IxNextgen()
443 ixnet_gen._ixnet = self.ixnet
444 with mock.patch.object(ixnet_gen, '_update_ipv4_address') as \
446 mock.patch.object(ixnet_gen, '_get_stack_item'), \
447 mock.patch.object(ixnet_gen,
448 '_get_config_element_by_flow_group_name', return_value='celm'):
449 ixnet_gen.update_ip_packet(TRAFFIC_PARAMETERS)
451 self.assertEqual(4, len(mock_update_add.mock_calls))
453 def test_update_ip_packet_exception_no_config_element(self):
454 ixnet_gen = ixnet_api.IxNextgen()
455 ixnet_gen._ixnet = self.ixnet
456 with mock.patch.object(ixnet_gen,
457 '_get_config_element_by_flow_group_name',
459 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
460 ixnet_gen.update_ip_packet(TRAFFIC_PARAMETERS)
462 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
463 def test_start_traffic(self, mock_ixnextgen_get_traffic_state):
464 ixnet_gen = ixnet_api.IxNextgen()
465 ixnet_gen._ixnet = self.ixnet
466 ixnet_gen._ixnet.getList.return_value = [0]
468 mock_ixnextgen_get_traffic_state.side_effect = [
469 'stopped', 'started', 'started', 'started']
471 result = ixnet_gen.start_traffic()
472 self.assertIsNone(result)
473 self.ixnet.getList.assert_called_once()
474 self.assertEqual(3, ixnet_gen._ixnet.execute.call_count)
476 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
477 def test_start_traffic_traffic_running(
478 self, mock_ixnextgen_get_traffic_state):
479 ixnet_gen = ixnet_api.IxNextgen()
480 ixnet_gen._ixnet = self.ixnet
481 ixnet_gen._ixnet.getList.return_value = [0]
482 mock_ixnextgen_get_traffic_state.side_effect = [
483 'started', 'stopped', 'started']
485 result = ixnet_gen.start_traffic()
486 self.assertIsNone(result)
487 self.ixnet.getList.assert_called_once()
488 self.assertEqual(4, ixnet_gen._ixnet.execute.call_count)
490 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
491 def test_start_traffic_wait_for_traffic_to_stop(
492 self, mock_ixnextgen_get_traffic_state):
493 ixnet_gen = ixnet_api.IxNextgen()
494 ixnet_gen._ixnet = self.ixnet
495 ixnet_gen._ixnet.getList.return_value = [0]
496 mock_ixnextgen_get_traffic_state.side_effect = [
497 'started', 'started', 'started', 'stopped', 'started']
499 result = ixnet_gen.start_traffic()
500 self.assertIsNone(result)
501 self.ixnet.getList.assert_called_once()
502 self.assertEqual(4, ixnet_gen._ixnet.execute.call_count)
504 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
505 def test_start_traffic_wait_for_traffic_start(
506 self, mock_ixnextgen_get_traffic_state):
507 ixnet_gen = ixnet_api.IxNextgen()
508 ixnet_gen._ixnet = self.ixnet
509 ixnet_gen._ixnet.getList.return_value = [0]
510 mock_ixnextgen_get_traffic_state.side_effect = [
511 'stopped', 'stopped', 'stopped', 'started']
513 result = ixnet_gen.start_traffic()
514 self.assertIsNone(result)
515 self.ixnet.getList.assert_called_once()
516 self.assertEqual(3, ixnet_gen._ixnet.execute.call_count)