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 copy import deepcopy
21 from yardstick.common import exceptions
22 from yardstick.network_services.libs.ixia_libs.ixnet import ixnet_api
28 TRAFFIC_PARAMETERS = {
36 'framesize': {'64B': '25', '256B': '75'},
45 'dstip': '152.16.40.20',
46 'srcip': '152.16.100.20',
58 'traffic_type': 'continuous'
67 'framesize': {'128B': '35', '1024B': '65'},
89 'traffic_type': 'continuous'
94 class TestIxNextgen(unittest.TestCase):
97 self.ixnet = mock.Mock()
98 self.ixnet.execute = mock.Mock()
99 self.ixnet.getRoot.return_value = 'my_root'
100 self.ixnet_gen = ixnet_api.IxNextgen()
101 self.ixnet_gen._ixnet = self.ixnet
103 def test_get_config(self):
107 'external-interface': [
108 {'virtual-interface': {'vpci': '0000:07:00.1'}},
109 {'virtual-interface': {'vpci': '0001:08:01.2'}}
116 'dut_result_dir': 'test2',
118 'ixchassis': 'test4',
128 'cards': ['0000', '0001'],
129 'ports': ['07', '08'],
130 'output_dir': 'test2',
135 result = ixnet_api.IxNextgen.get_config(tg_cfg)
136 self.assertEqual(result, expected)
138 def test__get_config_element_by_flow_group_name(self):
139 self.ixnet_gen._ixnet.getList.side_effect = [['traffic_item'],
141 self.ixnet_gen._ixnet.getAttribute.return_value = 'flow_group_01'
142 output = self.ixnet_gen._get_config_element_by_flow_group_name(
144 self.assertEqual('traffic_item/configElement:flow_group_01', output)
146 def test__get_config_element_by_flow_group_name_no_match(self):
147 self.ixnet_gen._ixnet.getList.side_effect = [['traffic_item'],
149 self.ixnet_gen._ixnet.getAttribute.return_value = 'flow_group_02'
150 output = self.ixnet_gen._get_config_element_by_flow_group_name(
152 self.assertIsNone(output)
154 def test__get_stack_item(self):
155 self.ixnet_gen._ixnet.getList.return_value = ['tcp1', 'tcp2', 'udp']
156 with mock.patch.object(
157 self.ixnet_gen, '_get_config_element_by_flow_group_name') as \
158 mock_get_cfg_element:
159 mock_get_cfg_element.return_value = 'cfg_element'
160 output = self.ixnet_gen._get_stack_item(mock.ANY, ixnet_api.PROTO_TCP)
161 self.assertEqual(['tcp1', 'tcp2'], output)
163 def test__get_stack_item_no_config_element(self):
164 with mock.patch.object(
165 self.ixnet_gen, '_get_config_element_by_flow_group_name',
167 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
168 self.ixnet_gen._get_stack_item(mock.ANY, mock.ANY)
170 def test__get_field_in_stack_item(self):
171 self.ixnet_gen._ixnet.getList.return_value = ['field1', 'field2']
172 output = self.ixnet_gen._get_field_in_stack_item(mock.ANY, 'field2')
173 self.assertEqual('field2', output)
175 def test__get_field_in_stack_item_no_field_present(self):
176 self.ixnet_gen._ixnet.getList.return_value = ['field1', 'field2']
177 with self.assertRaises(exceptions.IxNetworkFieldNotPresentInStackItem):
178 self.ixnet_gen._get_field_in_stack_item(mock.ANY, 'field3')
180 def test__parse_framesize(self):
181 framesize = {'64B': '75', '512b': '25'}
182 output = self.ixnet_gen._parse_framesize(framesize)
183 self.assertEqual(2, len(output))
184 self.assertIn([64, 64, 75], output)
185 self.assertIn([512, 512, 25], output)
187 @mock.patch.object(IxNetwork, 'IxNet')
188 def test_connect(self, mock_ixnet):
189 mock_ixnet.return_value = self.ixnet
190 with mock.patch.object(self.ixnet_gen, 'get_config') as mock_config:
191 mock_config.return_value = {'machine': 'machine_fake',
194 self.ixnet_gen.connect(mock.ANY)
196 self.ixnet.connect.assert_called_once_with(
197 'machine_fake', '-port', 'port_fake', '-version', '12345')
198 mock_config.assert_called_once()
200 def test_connect_invalid_config_no_machine(self):
201 self.ixnet_gen.get_config = mock.Mock(return_value={
204 self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY)
205 self.ixnet.connect.assert_not_called()
207 def test_connect_invalid_config_no_port(self):
208 self.ixnet_gen.get_config = mock.Mock(return_value={
209 'machine': 'machine_fake',
211 self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY)
212 self.ixnet.connect.assert_not_called()
214 def test_connect_invalid_config_no_version(self):
215 self.ixnet_gen.get_config = mock.Mock(return_value={
216 'machine': 'machine_fake',
217 'port': 'port_fake'})
218 self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY)
219 self.ixnet.connect.assert_not_called()
221 def test_connect_no_config(self):
222 self.ixnet_gen.get_config = mock.Mock(return_value={})
223 self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY)
224 self.ixnet.connect.assert_not_called()
226 def test_clear_config(self):
227 self.ixnet_gen.clear_config()
228 self.ixnet.execute.assert_called_once_with('newConfig')
230 @mock.patch.object(ixnet_api, 'log')
231 def test_assign_ports_2_ports(self, *args):
232 self.ixnet.getAttribute.side_effect = ['up', 'down']
234 'chassis': '1.1.1.1',
237 self.ixnet_gen._cfg = config
239 self.assertIsNone(self.ixnet_gen.assign_ports())
240 self.assertEqual(self.ixnet.execute.call_count, 2)
241 self.assertEqual(self.ixnet.commit.call_count, 4)
242 self.assertEqual(self.ixnet.getAttribute.call_count, 2)
244 @mock.patch.object(ixnet_api, 'log')
245 def test_assign_ports_port_down(self, mock_log):
246 self.ixnet.getAttribute.return_value = 'down'
248 'chassis': '1.1.1.1',
251 self.ixnet_gen._cfg = config
252 self.ixnet_gen.assign_ports()
253 mock_log.warning.assert_called()
255 def test_assign_ports_no_config(self):
256 self.ixnet_gen._cfg = {}
257 self.assertRaises(KeyError, self.ixnet_gen.assign_ports)
259 def test__create_traffic_item(self):
260 self.ixnet.add.return_value = 'my_new_traffic_item'
261 self.ixnet.remapIds.return_value = ['my_traffic_item_id']
263 self.ixnet_gen._create_traffic_item()
264 self.ixnet.add.assert_called_once_with(
265 'my_root/traffic', 'trafficItem')
266 self.ixnet.setMultiAttribute.assert_called_once_with(
267 'my_new_traffic_item', '-name', 'RFC2544', '-trafficType', 'raw')
268 self.assertEqual(2, self.ixnet.commit.call_count)
269 self.ixnet.remapIds.assert_called_once_with('my_new_traffic_item')
270 self.ixnet.setAttribute('my_traffic_item_id/tracking',
271 '-trackBy', 'trafficGroupId0')
273 def test__create_flow_groups(self):
274 self.ixnet_gen.ixnet.getList.side_effect = [['traffic_item'], ['1', '2']]
275 self.ixnet_gen.ixnet.add.side_effect = ['endp1', 'endp2']
276 self.ixnet_gen._create_flow_groups()
277 self.ixnet_gen.ixnet.add.assert_has_calls([
278 mock.call('traffic_item', 'endpointSet'),
279 mock.call('traffic_item', 'endpointSet')])
280 self.ixnet_gen.ixnet.setMultiAttribute.assert_has_calls([
281 mock.call('endp1', '-name', '1', '-sources', ['1/protocols'],
282 '-destinations', ['2/protocols']),
283 mock.call('endp2', '-name', '2', '-sources', ['2/protocols'],
284 '-destinations', ['1/protocols'])])
286 def test__append_protocol_to_stack(self):
288 self.ixnet_gen._append_procotol_to_stack('my_protocol', 'prev_element')
289 self.ixnet.execute.assert_called_with(
290 'append', 'prev_element',
291 'my_root/traffic/protocolTemplate:"my_protocol"')
293 def test__setup_config_elements(self):
294 self.ixnet_gen.ixnet.getList.side_effect = [['traffic_item'],
296 with mock.patch.object(self.ixnet_gen, '_append_procotol_to_stack') as \
298 self.ixnet_gen._setup_config_elements()
299 mock_append_proto.assert_has_calls([
300 mock.call(ixnet_api.PROTO_UDP, 'cfg_element/stack:"ethernet-1"'),
301 mock.call(ixnet_api.PROTO_IPV4, 'cfg_element/stack:"ethernet-1"')])
302 self.ixnet_gen.ixnet.setAttribute.assert_has_calls([
303 mock.call('cfg_element/frameRateDistribution', '-portDistribution',
305 mock.call('cfg_element/frameRateDistribution',
306 '-streamDistribution', 'splitRateEvenly')])
308 @mock.patch.object(ixnet_api.IxNextgen, '_create_traffic_item')
309 @mock.patch.object(ixnet_api.IxNextgen, '_create_flow_groups')
310 @mock.patch.object(ixnet_api.IxNextgen, '_setup_config_elements')
311 def test_create_traffic_model(self, mock__setup_config_elements,
312 mock__create_flow_groups,
313 mock__create_traffic_item):
315 self.ixnet_gen.create_traffic_model()
316 mock__create_traffic_item.assert_called_once()
317 mock__create_flow_groups.assert_called_once()
318 mock__setup_config_elements.assert_called_once()
320 def test__update_frame_mac(self):
321 with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item') as \
323 mock_get_field.return_value = 'field_descriptor'
324 self.ixnet_gen._update_frame_mac('ethernet_descriptor', 'field', 'mac')
325 mock_get_field.assert_called_once_with('ethernet_descriptor', 'field')
326 self.ixnet_gen.ixnet.setMultiAttribute(
327 'field_descriptor', '-singleValue', 'mac', '-fieldValue', 'mac',
328 '-valueType', 'singleValue')
329 self.ixnet_gen.ixnet.commit.assert_called_once()
331 def test_update_frame(self):
332 with mock.patch.object(
333 self.ixnet_gen, '_get_config_element_by_flow_group_name',
334 return_value='cfg_element'), \
335 mock.patch.object(self.ixnet_gen, '_update_frame_mac') as \
337 mock.patch.object(self.ixnet_gen, '_get_stack_item') as \
339 mock_get_stack_item.side_effect = [['item1'], ['item2'],
340 ['item3'], ['item4']]
341 self.ixnet_gen.update_frame(TRAFFIC_PARAMETERS, 50)
343 self.assertEqual(6, len(self.ixnet_gen.ixnet.setMultiAttribute.mock_calls))
344 self.assertEqual(4, len(mock_update_frame.mock_calls))
346 self.ixnet_gen.ixnet.setMultiAttribute.assert_has_calls([
347 mock.call('cfg_element/transmissionControl',
348 '-type', 'continuous', '-duration', 50)
351 def test_update_frame_qinq(self):
352 with mock.patch.object(self.ixnet_gen,
353 '_get_config_element_by_flow_group_name',
354 return_value='cfg_element'), \
355 mock.patch.object(self.ixnet_gen, '_update_frame_mac'),\
356 mock.patch.object(self.ixnet_gen, '_get_stack_item',
357 return_value='item'), \
358 mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item',
359 return_value='field'):
361 traffic_parameters = deepcopy(TRAFFIC_PARAMETERS)
362 traffic_parameters[UPLINK]['outer_l2']['QinQ'] = {
363 'S-VLAN': {'id': 128,
366 'C-VLAN': {'id': 512,
371 self.ixnet_gen.update_frame(traffic_parameters, 50)
373 self.ixnet_gen.ixnet.setMultiAttribute.assert_has_calls([
374 mock.call('field', '-auto', 'false', '-singleValue', '0x88a8',
375 '-fieldValue', '0x88a8', '-valueType', 'singleValue'),
376 mock.call('field', '-auto', 'false', '-singleValue', 1,
377 '-fieldValue', 1, '-valueType', 'singleValue'),
378 mock.call('field', '-auto', 'false', '-singleValue', 128,
379 '-fieldValue', 128, '-valueType', 'singleValue'),
380 mock.call('field', '-auto', 'false', '-singleValue', 512,
381 '-fieldValue', 512, '-valueType', 'singleValue'),
382 mock.call('field', '-auto', 'false', '-singleValue', 2,
383 '-fieldValue', 2, '-valueType', 'singleValue')
386 def test_update_frame_flow_not_present(self):
387 with mock.patch.object(
388 self.ixnet_gen, '_get_config_element_by_flow_group_name',
390 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
391 self.ixnet_gen.update_frame(TRAFFIC_PARAMETERS, 40)
393 def test_get_statistics(self):
394 port_statistics = '::ixNet::OBJ-/statistics/view:"Port Statistics"'
395 flow_statistics = '::ixNet::OBJ-/statistics/view:"Flow Statistics"'
396 with mock.patch.object(self.ixnet_gen, '_build_stats_map') as \
398 self.ixnet_gen.get_statistics()
400 mock_build_stats.assert_has_calls([
401 mock.call(port_statistics, self.ixnet_gen.PORT_STATS_NAME_MAP),
402 mock.call(flow_statistics, self.ixnet_gen.LATENCY_NAME_MAP)])
404 def test__update_ipv4_address(self):
405 with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item',
406 return_value='field_desc'):
407 self.ixnet_gen._update_ipv4_address(mock.ANY, mock.ANY, '192.168.1.1',
409 self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with(
410 'field_desc', '-seed', 100, '-fixedBits', '192.168.1.1',
411 '-randomMask', '0.0.0.63', '-valueType', 'random',
414 def test__update_udp_port(self):
415 with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item',
416 return_value='field_desc'):
417 self.ixnet_gen._update_udp_port(mock.ANY, mock.ANY, 1234,
420 self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with(
426 '-valueType', 'random',
429 def test_update_ip_packet(self):
430 with mock.patch.object(self.ixnet_gen, '_update_ipv4_address') as \
432 mock.patch.object(self.ixnet_gen, '_get_stack_item'), \
433 mock.patch.object(self.ixnet_gen,
434 '_get_config_element_by_flow_group_name', return_value='celm'):
435 self.ixnet_gen.update_ip_packet(TRAFFIC_PARAMETERS)
437 self.assertEqual(4, len(mock_update_add.mock_calls))
439 def test_update_ip_packet_exception_no_config_element(self):
440 with mock.patch.object(self.ixnet_gen,
441 '_get_config_element_by_flow_group_name',
443 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
444 self.ixnet_gen.update_ip_packet(TRAFFIC_PARAMETERS)
446 def test_update_l4(self):
447 with mock.patch.object(self.ixnet_gen, '_update_udp_port') as \
449 mock.patch.object(self.ixnet_gen, '_get_stack_item'), \
450 mock.patch.object(self.ixnet_gen,
451 '_get_config_element_by_flow_group_name', return_value='celm'):
452 self.ixnet_gen.update_l4(TRAFFIC_PARAMETERS)
454 self.assertEqual(4, len(mock_update_udp.mock_calls))
456 def test_update_l4_exception_no_config_element(self):
457 with mock.patch.object(self.ixnet_gen,
458 '_get_config_element_by_flow_group_name',
460 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
461 self.ixnet_gen.update_l4(TRAFFIC_PARAMETERS)
463 def test_update_l4_exception_no_supported_proto(self):
464 traffic_parameters = {
468 'proto': 'unsupported',
472 with mock.patch.object(self.ixnet_gen,
473 '_get_config_element_by_flow_group_name',
474 return_value='celm'):
475 with self.assertRaises(exceptions.IXIAUnsupportedProtocol):
476 self.ixnet_gen.update_l4(traffic_parameters)
478 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
479 def test_start_traffic(self, mock_ixnextgen_get_traffic_state):
480 self.ixnet_gen._ixnet.getList.return_value = [0]
482 mock_ixnextgen_get_traffic_state.side_effect = [
483 'stopped', 'started', 'started', 'started']
485 result = self.ixnet_gen.start_traffic()
486 self.assertIsNone(result)
487 self.ixnet.getList.assert_called_once()
488 self.assertEqual(3, self.ixnet_gen._ixnet.execute.call_count)
490 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
491 def test_start_traffic_traffic_running(
492 self, mock_ixnextgen_get_traffic_state):
493 self.ixnet_gen._ixnet.getList.return_value = [0]
494 mock_ixnextgen_get_traffic_state.side_effect = [
495 'started', 'stopped', 'started']
497 result = self.ixnet_gen.start_traffic()
498 self.assertIsNone(result)
499 self.ixnet.getList.assert_called_once()
500 self.assertEqual(4, self.ixnet_gen._ixnet.execute.call_count)
502 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
503 def test_start_traffic_wait_for_traffic_to_stop(
504 self, mock_ixnextgen_get_traffic_state):
505 self.ixnet_gen._ixnet.getList.return_value = [0]
506 mock_ixnextgen_get_traffic_state.side_effect = [
507 'started', 'started', 'started', 'stopped', 'started']
509 result = self.ixnet_gen.start_traffic()
510 self.assertIsNone(result)
511 self.ixnet.getList.assert_called_once()
512 self.assertEqual(4, self.ixnet_gen._ixnet.execute.call_count)
514 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
515 def test_start_traffic_wait_for_traffic_start(
516 self, mock_ixnextgen_get_traffic_state):
517 self.ixnet_gen._ixnet.getList.return_value = [0]
518 mock_ixnextgen_get_traffic_state.side_effect = [
519 'stopped', 'stopped', 'stopped', 'started']
521 result = self.ixnet_gen.start_traffic()
522 self.assertIsNone(result)
523 self.ixnet.getList.assert_called_once()
524 self.assertEqual(3, self.ixnet_gen._ixnet.execute.call_count)