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'}
41 'dstip': '152.16.40.20',
42 'srcip': '152.16.100.20',
54 'traffic_type': 'continuous'
63 'framesize': {'128B': '35', '1024B': '65'}
83 'traffic_type': 'continuous'
88 class TestIxNextgen(unittest.TestCase):
91 self.ixnet = mock.Mock()
92 self.ixnet.execute = mock.Mock()
93 self.ixnet.getRoot.return_value = 'my_root'
94 self.ixnet_gen = ixnet_api.IxNextgen()
95 self.ixnet_gen._ixnet = self.ixnet
97 def test_get_config(self):
101 'external-interface': [
102 {'virtual-interface': {'vpci': '0000:07:00.1'}},
103 {'virtual-interface': {'vpci': '0001:08:01.2'}}
110 'dut_result_dir': 'test2',
112 'ixchassis': 'test4',
122 'cards': ['0000', '0001'],
123 'ports': ['07', '08'],
124 'output_dir': 'test2',
129 result = ixnet_api.IxNextgen.get_config(tg_cfg)
130 self.assertEqual(result, expected)
132 def test__get_config_element_by_flow_group_name(self):
133 self.ixnet_gen._ixnet.getList.side_effect = [['traffic_item'],
135 self.ixnet_gen._ixnet.getAttribute.return_value = 'flow_group_01'
136 output = self.ixnet_gen._get_config_element_by_flow_group_name(
138 self.assertEqual('traffic_item/configElement:flow_group_01', output)
140 def test__get_config_element_by_flow_group_name_no_match(self):
141 self.ixnet_gen._ixnet.getList.side_effect = [['traffic_item'],
143 self.ixnet_gen._ixnet.getAttribute.return_value = 'flow_group_02'
144 output = self.ixnet_gen._get_config_element_by_flow_group_name(
146 self.assertIsNone(output)
148 def test__get_stack_item(self):
149 self.ixnet_gen._ixnet.getList.return_value = ['tcp1', 'tcp2', 'udp']
150 with mock.patch.object(
151 self.ixnet_gen, '_get_config_element_by_flow_group_name') as \
152 mock_get_cfg_element:
153 mock_get_cfg_element.return_value = 'cfg_element'
154 output = self.ixnet_gen._get_stack_item(mock.ANY, ixnet_api.PROTO_TCP)
155 self.assertEqual(['tcp1', 'tcp2'], output)
157 def test__get_stack_item_no_config_element(self):
158 with mock.patch.object(
159 self.ixnet_gen, '_get_config_element_by_flow_group_name',
161 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
162 self.ixnet_gen._get_stack_item(mock.ANY, mock.ANY)
164 def test__get_field_in_stack_item(self):
165 self.ixnet_gen._ixnet.getList.return_value = ['field1', 'field2']
166 output = self.ixnet_gen._get_field_in_stack_item(mock.ANY, 'field2')
167 self.assertEqual('field2', output)
169 def test__get_field_in_stack_item_no_field_present(self):
170 self.ixnet_gen._ixnet.getList.return_value = ['field1', 'field2']
171 with self.assertRaises(exceptions.IxNetworkFieldNotPresentInStackItem):
172 self.ixnet_gen._get_field_in_stack_item(mock.ANY, 'field3')
174 def test__parse_framesize(self):
175 framesize = {'64B': '75', '512b': '25'}
176 output = self.ixnet_gen._parse_framesize(framesize)
177 self.assertEqual(2, len(output))
178 self.assertIn([64, 64, 75], output)
179 self.assertIn([512, 512, 25], output)
181 @mock.patch.object(IxNetwork, 'IxNet')
182 def test_connect(self, mock_ixnet):
183 mock_ixnet.return_value = self.ixnet
184 with mock.patch.object(self.ixnet_gen, 'get_config') as mock_config:
185 mock_config.return_value = {'machine': 'machine_fake',
188 self.ixnet_gen.connect(mock.ANY)
190 self.ixnet.connect.assert_called_once_with(
191 'machine_fake', '-port', 'port_fake', '-version', '12345')
192 mock_config.assert_called_once()
194 def test_connect_invalid_config_no_machine(self):
195 self.ixnet_gen.get_config = mock.Mock(return_value={
198 self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY)
199 self.ixnet.connect.assert_not_called()
201 def test_connect_invalid_config_no_port(self):
202 self.ixnet_gen.get_config = mock.Mock(return_value={
203 'machine': 'machine_fake',
205 self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY)
206 self.ixnet.connect.assert_not_called()
208 def test_connect_invalid_config_no_version(self):
209 self.ixnet_gen.get_config = mock.Mock(return_value={
210 'machine': 'machine_fake',
211 'port': 'port_fake'})
212 self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY)
213 self.ixnet.connect.assert_not_called()
215 def test_connect_no_config(self):
216 self.ixnet_gen.get_config = mock.Mock(return_value={})
217 self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY)
218 self.ixnet.connect.assert_not_called()
220 def test_clear_config(self):
221 self.ixnet_gen.clear_config()
222 self.ixnet.execute.assert_called_once_with('newConfig')
224 @mock.patch.object(ixnet_api, 'log')
225 def test_assign_ports_2_ports(self, *args):
226 self.ixnet.getAttribute.side_effect = ['up', 'down']
228 'chassis': '1.1.1.1',
231 self.ixnet_gen._cfg = config
233 self.assertIsNone(self.ixnet_gen.assign_ports())
234 self.assertEqual(self.ixnet.execute.call_count, 2)
235 self.assertEqual(self.ixnet.commit.call_count, 4)
236 self.assertEqual(self.ixnet.getAttribute.call_count, 2)
238 @mock.patch.object(ixnet_api, 'log')
239 def test_assign_ports_port_down(self, mock_log):
240 self.ixnet.getAttribute.return_value = 'down'
242 'chassis': '1.1.1.1',
245 self.ixnet_gen._cfg = config
246 self.ixnet_gen.assign_ports()
247 mock_log.warning.assert_called()
249 def test_assign_ports_no_config(self):
250 self.ixnet_gen._cfg = {}
251 self.assertRaises(KeyError, self.ixnet_gen.assign_ports)
253 def test__create_traffic_item(self):
254 self.ixnet.add.return_value = 'my_new_traffic_item'
255 self.ixnet.remapIds.return_value = ['my_traffic_item_id']
257 self.ixnet_gen._create_traffic_item()
258 self.ixnet.add.assert_called_once_with(
259 'my_root/traffic', 'trafficItem')
260 self.ixnet.setMultiAttribute.assert_called_once_with(
261 'my_new_traffic_item', '-name', 'RFC2544', '-trafficType', 'raw')
262 self.assertEqual(2, self.ixnet.commit.call_count)
263 self.ixnet.remapIds.assert_called_once_with('my_new_traffic_item')
264 self.ixnet.setAttribute('my_traffic_item_id/tracking',
265 '-trackBy', 'trafficGroupId0')
267 def test__create_flow_groups(self):
268 self.ixnet_gen.ixnet.getList.side_effect = [['traffic_item'], ['1', '2']]
269 self.ixnet_gen.ixnet.add.side_effect = ['endp1', 'endp2']
270 self.ixnet_gen._create_flow_groups()
271 self.ixnet_gen.ixnet.add.assert_has_calls([
272 mock.call('traffic_item', 'endpointSet'),
273 mock.call('traffic_item', 'endpointSet')])
274 self.ixnet_gen.ixnet.setMultiAttribute.assert_has_calls([
275 mock.call('endp1', '-name', '1', '-sources', ['1/protocols'],
276 '-destinations', ['2/protocols']),
277 mock.call('endp2', '-name', '2', '-sources', ['2/protocols'],
278 '-destinations', ['1/protocols'])])
280 def test__append_protocol_to_stack(self):
282 self.ixnet_gen._append_procotol_to_stack('my_protocol', 'prev_element')
283 self.ixnet.execute.assert_called_with(
284 'append', 'prev_element',
285 'my_root/traffic/protocolTemplate:"my_protocol"')
287 def test__setup_config_elements(self):
288 self.ixnet_gen.ixnet.getList.side_effect = [['traffic_item'],
290 with mock.patch.object(self.ixnet_gen, '_append_procotol_to_stack') as \
292 self.ixnet_gen._setup_config_elements()
293 mock_append_proto.assert_has_calls([
294 mock.call(ixnet_api.PROTO_UDP, 'cfg_element/stack:"ethernet-1"'),
295 mock.call(ixnet_api.PROTO_IPV4, 'cfg_element/stack:"ethernet-1"')])
296 self.ixnet_gen.ixnet.setAttribute.assert_has_calls([
297 mock.call('cfg_element/frameRateDistribution', '-portDistribution',
299 mock.call('cfg_element/frameRateDistribution',
300 '-streamDistribution', 'splitRateEvenly')])
302 @mock.patch.object(ixnet_api.IxNextgen, '_create_traffic_item')
303 @mock.patch.object(ixnet_api.IxNextgen, '_create_flow_groups')
304 @mock.patch.object(ixnet_api.IxNextgen, '_setup_config_elements')
305 def test_create_traffic_model(self, mock__setup_config_elements,
306 mock__create_flow_groups,
307 mock__create_traffic_item):
309 self.ixnet_gen.create_traffic_model()
310 mock__create_traffic_item.assert_called_once()
311 mock__create_flow_groups.assert_called_once()
312 mock__setup_config_elements.assert_called_once()
314 def test__update_frame_mac(self):
315 with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item') as \
317 mock_get_field.return_value = 'field_descriptor'
318 self.ixnet_gen._update_frame_mac('ethernet_descriptor', 'field', 'mac')
319 mock_get_field.assert_called_once_with('ethernet_descriptor', 'field')
320 self.ixnet_gen.ixnet.setMultiAttribute(
321 'field_descriptor', '-singleValue', 'mac', '-fieldValue', 'mac',
322 '-valueType', 'singleValue')
323 self.ixnet_gen.ixnet.commit.assert_called_once()
325 def test_update_frame(self):
326 with mock.patch.object(
327 self.ixnet_gen, '_get_config_element_by_flow_group_name',
328 return_value='cfg_element'), \
329 mock.patch.object(self.ixnet_gen, '_update_frame_mac') as \
331 mock.patch.object(self.ixnet_gen, '_get_stack_item') as \
333 mock_get_stack_item.side_effect = [['item1'], ['item2'],
334 ['item3'], ['item4']]
335 self.ixnet_gen.update_frame(TRAFFIC_PARAMETERS, 50)
337 self.assertEqual(6, len(self.ixnet_gen.ixnet.setMultiAttribute.mock_calls))
338 self.assertEqual(4, len(mock_update_frame.mock_calls))
340 def test_update_frame_flow_not_present(self):
341 with mock.patch.object(
342 self.ixnet_gen, '_get_config_element_by_flow_group_name',
344 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
345 self.ixnet_gen.update_frame(TRAFFIC_PARAMETERS, 40)
347 def test_get_statistics(self):
348 port_statistics = '::ixNet::OBJ-/statistics/view:"Port Statistics"'
349 flow_statistics = '::ixNet::OBJ-/statistics/view:"Flow Statistics"'
350 with mock.patch.object(self.ixnet_gen, '_build_stats_map') as \
352 self.ixnet_gen.get_statistics()
354 mock_build_stats.assert_has_calls([
355 mock.call(port_statistics, self.ixnet_gen.PORT_STATS_NAME_MAP),
356 mock.call(flow_statistics, self.ixnet_gen.LATENCY_NAME_MAP)])
358 def test__update_ipv4_address(self):
359 with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item',
360 return_value='field_desc'):
361 self.ixnet_gen._update_ipv4_address(mock.ANY, mock.ANY, '192.168.1.1',
363 self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with(
364 'field_desc', '-seed', 100, '-fixedBits', '192.168.1.1',
365 '-randomMask', '0.0.0.63', '-valueType', 'random',
368 def test__update_udp_port(self):
369 with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item',
370 return_value='field_desc'):
371 self.ixnet_gen._update_udp_port(mock.ANY, mock.ANY, 1234,
374 self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with(
380 '-valueType', 'random',
383 def test_update_ip_packet(self):
384 with mock.patch.object(self.ixnet_gen, '_update_ipv4_address') as \
386 mock.patch.object(self.ixnet_gen, '_get_stack_item'), \
387 mock.patch.object(self.ixnet_gen,
388 '_get_config_element_by_flow_group_name', return_value='celm'):
389 self.ixnet_gen.update_ip_packet(TRAFFIC_PARAMETERS)
391 self.assertEqual(4, len(mock_update_add.mock_calls))
393 def test_update_ip_packet_exception_no_config_element(self):
394 with mock.patch.object(self.ixnet_gen,
395 '_get_config_element_by_flow_group_name',
397 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
398 self.ixnet_gen.update_ip_packet(TRAFFIC_PARAMETERS)
400 def test_update_l4(self):
401 with mock.patch.object(self.ixnet_gen, '_update_udp_port') as \
403 mock.patch.object(self.ixnet_gen, '_get_stack_item'), \
404 mock.patch.object(self.ixnet_gen,
405 '_get_config_element_by_flow_group_name', return_value='celm'):
406 self.ixnet_gen.update_l4(TRAFFIC_PARAMETERS)
408 self.assertEqual(4, len(mock_update_udp.mock_calls))
410 def test_update_l4_exception_no_config_element(self):
411 with mock.patch.object(self.ixnet_gen,
412 '_get_config_element_by_flow_group_name',
414 with self.assertRaises(exceptions.IxNetworkFlowNotPresent):
415 self.ixnet_gen.update_l4(TRAFFIC_PARAMETERS)
417 def test_update_l4_exception_no_supported_proto(self):
418 traffic_parameters = {
422 'proto': 'unsupported',
426 with mock.patch.object(self.ixnet_gen,
427 '_get_config_element_by_flow_group_name',
428 return_value='celm'):
429 with self.assertRaises(exceptions.IXIAUnsupportedProtocol):
430 self.ixnet_gen.update_l4(traffic_parameters)
432 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
433 def test_start_traffic(self, mock_ixnextgen_get_traffic_state):
434 self.ixnet_gen._ixnet.getList.return_value = [0]
436 mock_ixnextgen_get_traffic_state.side_effect = [
437 'stopped', 'started', 'started', 'started']
439 result = self.ixnet_gen.start_traffic()
440 self.assertIsNone(result)
441 self.ixnet.getList.assert_called_once()
442 self.assertEqual(3, self.ixnet_gen._ixnet.execute.call_count)
444 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
445 def test_start_traffic_traffic_running(
446 self, mock_ixnextgen_get_traffic_state):
447 self.ixnet_gen._ixnet.getList.return_value = [0]
448 mock_ixnextgen_get_traffic_state.side_effect = [
449 'started', 'stopped', 'started']
451 result = self.ixnet_gen.start_traffic()
452 self.assertIsNone(result)
453 self.ixnet.getList.assert_called_once()
454 self.assertEqual(4, self.ixnet_gen._ixnet.execute.call_count)
456 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
457 def test_start_traffic_wait_for_traffic_to_stop(
458 self, mock_ixnextgen_get_traffic_state):
459 self.ixnet_gen._ixnet.getList.return_value = [0]
460 mock_ixnextgen_get_traffic_state.side_effect = [
461 'started', 'started', 'started', 'stopped', 'started']
463 result = self.ixnet_gen.start_traffic()
464 self.assertIsNone(result)
465 self.ixnet.getList.assert_called_once()
466 self.assertEqual(4, self.ixnet_gen._ixnet.execute.call_count)
468 @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state')
469 def test_start_traffic_wait_for_traffic_start(
470 self, mock_ixnextgen_get_traffic_state):
471 self.ixnet_gen._ixnet.getList.return_value = [0]
472 mock_ixnextgen_get_traffic_state.side_effect = [
473 'stopped', 'stopped', 'stopped', 'started']
475 result = self.ixnet_gen.start_traffic()
476 self.assertIsNone(result)
477 self.ixnet.getList.assert_called_once()
478 self.assertEqual(3, self.ixnet_gen._ixnet.execute.call_count)