1 # Copyright (c) 2016-2017 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.
21 from collections import OrderedDict
23 from yardstick.common import utils
24 from yardstick.common import exceptions
25 from yardstick.benchmark import contexts
26 from yardstick.benchmark.contexts import base as ctx_base
27 from yardstick.network_services.libs.ixia_libs.ixnet import ixnet_api
28 from yardstick.network_services.traffic_profile import base as tp_base
29 from yardstick.network_services.vnf_generic.vnf import tg_rfc2544_ixia
32 TEST_FILE_YAML = 'nsb_test_case.yaml'
37 class TestIxiaResourceHelper(unittest.TestCase):
40 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
41 self.mock_IxNextgen = self._mock_IxNextgen.start()
42 self.addCleanup(self._stop_mocks)
44 def _stop_mocks(self):
45 self._mock_IxNextgen.stop()
47 def test___init___with_custom_rfc_helper(self):
48 class MyRfcHelper(tg_rfc2544_ixia.IxiaRfc2544Helper):
51 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(
52 mock.Mock(), MyRfcHelper)
53 self.assertIsInstance(ixia_resource_helper.rfc_helper, MyRfcHelper)
55 def test__init_ix_scenario(self):
56 mock_scenario = mock.Mock()
57 mock_scenario_helper = mock.Mock()
58 mock_scenario_helper.scenario_cfg = {'ixia_config': 'TestScenario',
59 'options': 'scenario_options'}
60 mock_setup_helper = mock.Mock(scenario_helper=mock_scenario_helper)
61 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock_setup_helper)
62 ixia_resource_helper._ixia_scenarios = {'TestScenario': mock_scenario}
63 ixia_resource_helper.client = 'client'
64 ixia_resource_helper.context_cfg = 'context'
65 ixia_resource_helper._init_ix_scenario()
66 mock_scenario.assert_called_once_with('client', 'context', 'scenario_options')
68 def test__init_ix_scenario_not_supported_cfg_type(self):
69 mock_scenario_helper = mock.Mock()
70 mock_scenario_helper.scenario_cfg = {'ixia_config': 'FakeScenario',
71 'options': 'scenario_options'}
72 mock_setup_helper = mock.Mock(scenario_helper=mock_scenario_helper)
73 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock_setup_helper)
74 ixia_resource_helper._ixia_scenarios = {'TestScenario': mock.Mock()}
75 with self.assertRaises(RuntimeError):
76 ixia_resource_helper._init_ix_scenario()
78 @mock.patch.object(tg_rfc2544_ixia.IxiaResourceHelper, '_init_ix_scenario')
79 def test_setup(self, mock__init_ix_scenario):
80 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
81 ixia_resource_helper.setup()
82 mock__init_ix_scenario.assert_called_once()
84 def test_stop_collect_with_client(self):
85 mock_client = mock.Mock()
86 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
87 ixia_resource_helper.client = mock_client
88 ixia_resource_helper._ix_scenario = mock.Mock()
89 ixia_resource_helper.stop_collect()
90 self.assertEqual(1, ixia_resource_helper._terminated.value)
91 ixia_resource_helper._ix_scenario.stop_protocols.assert_called_once()
93 def test_run_traffic(self):
94 mock_tprofile = mock.Mock()
95 mock_tprofile.config.duration = 10
96 mock_tprofile.get_drop_percentage.return_value = True, 'fake_samples'
97 ixia_rhelper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
98 ixia_rhelper.rfc_helper = mock.Mock()
99 ixia_rhelper.vnfd_helper = mock.Mock()
100 ixia_rhelper._ix_scenario = mock.Mock()
101 ixia_rhelper.vnfd_helper.port_pairs.all_ports = []
102 with mock.patch.object(ixia_rhelper, 'generate_samples'), \
103 mock.patch.object(ixia_rhelper, '_build_ports'), \
104 mock.patch.object(ixia_rhelper, '_initialize_client'), \
105 mock.patch.object(utils, 'wait_until_true'):
106 ixia_rhelper.run_traffic(mock_tprofile)
108 self.assertEqual('fake_samples', ixia_rhelper._queue.get())
109 mock_tprofile.update_traffic_profile.assert_called_once()
112 @mock.patch.object(tg_rfc2544_ixia, 'ixnet_api')
113 class TestIXIATrafficGen(unittest.TestCase):
114 VNFD = {'vnfd:vnfd-catalog':
116 [{'short-name': 'VpeVnf',
119 [{'network': '152.16.100.20',
120 'netmask': '255.255.255.0',
121 'gateway': '152.16.100.20',
123 {'network': '152.16.40.20',
124 'netmask': '255.255.255.0',
125 'gateway': '152.16.40.20',
127 'description': 'VPE approximation using DPDK',
128 'name': 'vpevnf-baremetal',
130 [{'network': '0064:ff9b:0:0:0:0:9810:6414',
132 'gateway': '0064:ff9b:0:0:0:0:9810:6414',
134 {'network': '0064:ff9b:0:0:0:0:9810:2814',
136 'gateway': '0064:ff9b:0:0:0:0:9810:2814',
138 'id': 'vpevnf-baremetal',
139 'external-interface':
140 [{'virtual-interface':
141 {'dst_mac': '00:00:00:00:00:04',
142 'vpci': '0000:05:00.0',
143 'local_ip': '152.16.100.19',
144 'type': 'PCI-PASSTHROUGH',
145 'netmask': '255.255.255.0',
147 'bandwidth': '10 Gbps',
149 'dst_ip': '152.16.100.20',
150 'local_iface_name': 'xe0',
151 'local_mac': '00:00:00:00:00:02'},
152 'vnfd-connection-point-ref': 'xe0',
154 {'virtual-interface':
155 {'dst_mac': '00:00:00:00:00:03',
156 'vpci': '0000:05:00.1',
157 'local_ip': '152.16.40.19',
158 'type': 'PCI-PASSTHROUGH',
160 'netmask': '255.255.255.0',
162 'bandwidth': '10 Gbps',
163 'dst_ip': '152.16.40.20',
164 'local_iface_name': 'xe1',
165 'local_mac': '00:00:00:00:00:01'},
166 'vnfd-connection-point-ref': 'xe1',
168 'description': 'Vpe approximation using DPDK',
170 {'vdu-id': 'vpevnf-baremetal',
176 {'kpi': ['packets_in', 'packets_fwd',
178 'connection-point': [{'type': 'VPORT', 'name': 'xe0'},
179 {'type': 'VPORT', 'name': 'xe1'}],
180 'id': 'VpeApproxVnf', 'name': 'VPEVnfSsh'}]}}
183 "schema": "isb:traffic_profile:0.1",
185 "description": "Fixed traffic profile to run UDP traffic",
187 "traffic_type": "FixedTraffic",
188 "frame_rate": 100, # pps
192 TC_YAML = {'scenarios': [{'tc_options':
193 {'rfc2544': {'allowed_drop_rate': '0.8 - 1'}},
194 'runner': {'duration': 400,
195 'interval': 35, 'type': 'Duration'},
197 {'flow': 'ipv4_1flow_Packets_vpe.yaml',
198 'imix': 'imix_voice.yaml'},
199 'vnf_options': {'vpe': {'cfg': 'vpe_config'}},
200 'traffic_profile': 'ipv4_throughput_vpe.yaml',
202 'nodes': {'tg__1': 'trafficgen_1.yardstick',
203 'vnf__1': 'vnf.yardstick'},
204 'topology': 'vpe_vnf_topology.yaml'}],
205 'context': {'nfvi_type': 'baremetal',
206 'type': contexts.CONTEXT_NODE,
208 'file': '/etc/yardstick/nodes/pod.yaml'},
209 'schema': 'yardstick:task:0.1'}
211 def test___init__(self, *args):
212 with mock.patch("yardstick.ssh.SSH") as ssh:
213 ssh_mock = mock.Mock(autospec=ssh.SSH)
215 mock.Mock(return_value=(0, "", ""))
216 ssh.from_node.return_value = ssh_mock
217 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
218 # NOTE(ralonsoh): check the object returned.
219 tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd, 'task_id')
221 def test_listen_traffic(self, *args):
222 with mock.patch("yardstick.ssh.SSH") as ssh:
223 ssh_mock = mock.Mock(autospec=ssh.SSH)
225 mock.Mock(return_value=(0, "", ""))
226 ssh.from_node.return_value = ssh_mock
227 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
228 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd,
230 self.assertIsNone(ixnet_traffic_gen.listen_traffic({}))
232 @mock.patch.object(ctx_base.Context, 'get_context_from_server', return_value='fake_context')
233 def test_instantiate(self, *args):
234 with mock.patch("yardstick.ssh.SSH") as ssh:
235 ssh_mock = mock.Mock(autospec=ssh.SSH)
237 mock.Mock(return_value=(0, "", ""))
239 mock.Mock(return_value=(0, "", ""))
240 ssh.from_node.return_value = ssh_mock
241 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
242 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd,
244 scenario_cfg = {'tc': "nsb_test_case",
252 'allowed_drop_rate': '0.8 - 1'},
254 'rules': 'acl_1rule.yaml',
258 'worker_config': '1C/1T',
259 'worker_threads': 1}}}})
260 scenario_cfg.update({
261 'nodes': {ixnet_traffic_gen.name: "mock"}
263 ixnet_traffic_gen.topology = ""
264 ixnet_traffic_gen.get_ixobj = mock.MagicMock()
265 ixnet_traffic_gen._ixia_traffic_gen = mock.MagicMock()
266 ixnet_traffic_gen._ixia_traffic_gen._connect = mock.Mock()
269 ixnet_traffic_gen.instantiate(scenario_cfg, {}))
271 @mock.patch.object(ctx_base.Context, 'get_physical_node_from_server', return_value='mock_node')
272 def test_collect_kpi(self, *args):
273 with mock.patch("yardstick.ssh.SSH") as ssh:
274 ssh_mock = mock.Mock(autospec=ssh.SSH)
276 mock.Mock(return_value=(0, "", ""))
277 ssh.from_node.return_value = ssh_mock
279 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
280 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd,
282 ixnet_traffic_gen.scenario_helper.scenario_cfg = {
283 'nodes': {ixnet_traffic_gen.name: "mock"}
285 ixnet_traffic_gen.data = {}
286 restult = ixnet_traffic_gen.collect_kpi()
288 expected = {'collect_stats': {},
289 'physical_node': 'mock_node'}
291 self.assertEqual(expected, restult)
293 def test_terminate(self, *args):
294 with mock.patch("yardstick.ssh.SSH") as ssh:
295 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
296 ssh_mock = mock.Mock(autospec=ssh.SSH)
298 mock.Mock(return_value=(0, "", ""))
299 ssh.from_node.return_value = ssh_mock
300 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(
301 NAME, vnfd, 'task_id', resource_helper_type=mock.Mock())
302 ixnet_traffic_gen._terminated = mock.MagicMock()
303 ixnet_traffic_gen._terminated.value = 0
304 ixnet_traffic_gen._ixia_traffic_gen = mock.MagicMock()
305 ixnet_traffic_gen._ixia_traffic_gen.ix_stop_traffic = mock.Mock()
306 ixnet_traffic_gen._traffic_process = mock.MagicMock()
307 ixnet_traffic_gen._traffic_process.terminate = mock.Mock()
308 self.assertIsNone(ixnet_traffic_gen.terminate())
310 def _get_file_abspath(self, filename):
311 curr_path = os.path.dirname(os.path.abspath(__file__))
312 file_path = os.path.join(curr_path, filename)
315 def test__check_status(self, *args):
316 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
317 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd, 'task_id')
320 @mock.patch("yardstick.ssh.SSH")
321 def test_traffic_runner(self, mock_ssh, *args):
322 mock_traffic_profile = mock.Mock(autospec=tp_base.TrafficProfile)
323 mock_traffic_profile.get_traffic_definition.return_value = "64"
324 mock_traffic_profile.params = self.TRAFFIC_PROFILE
325 # traffic_profile.ports is standardized on port_num
326 mock_traffic_profile.ports = [0, 1]
328 mock_ssh_instance = mock.Mock(autospec=mock_ssh.SSH)
329 mock_ssh_instance.execute.return_value = 0, "", ""
330 mock_ssh_instance.run.return_value = 0, "", ""
332 mock_ssh.from_node.return_value = mock_ssh_instance
334 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
335 vnfd["mgmt-interface"].update({
337 "ixchassis": "1.1.1.1",
338 "py_bin_path": "/root",
344 for ifname in range(1):
345 name = "xe{}".format(ifname)
353 "Valid_Frames_Rx": 1000,
359 samples.update({"CurrentDropPercentage": 0.0})
364 "Rx_Rate_Kbps": [20, 20],
365 "Tx_Rate_Kbps": [20, 20],
366 "Rx_Rate_Mbps": [10, 10],
367 "Tx_Rate_Mbps": [10, 10],
368 "CurrentDropPercentage": [0, 0],
369 "RxThroughput": [10, 10],
370 "TxThroughput": [10, 10],
371 "Frames_Tx": [1000, 1000],
372 "in_packets": [1000, 1000],
373 "Valid_Frames_Rx": [1000, 1000],
374 "out_packets": [1000, 1000],
378 mock_traffic_profile.execute_traffic.return_value = [
379 'Completed', samples]
380 mock_traffic_profile.get_drop_percentage.return_value = [
381 'Completed', samples]
383 sut = tg_rfc2544_ixia.IxiaTrafficGen(name, vnfd, 'task_id')
384 sut.vnf_port_pairs = [[[0], [1]]]
385 sut.tc_file_name = self._get_file_abspath(TEST_FILE_YAML)
388 sut.ssh_helper = mock.Mock()
389 sut._traffic_process = mock.MagicMock()
390 sut.generate_port_pairs = mock.Mock()
392 sut._ixia_traffic_gen = mock.MagicMock()
393 sut._ixia_traffic_gen.ix_get_statistics.return_value = last_res
395 sut.resource_helper.client = mock.MagicMock()
396 sut.resource_helper.client_started = mock.MagicMock()
397 sut.resource_helper.client_started.value = 1
398 sut.resource_helper.rfc_helper.iteration.value = 11
399 sut.resource_helper._ix_scenario = mock.Mock()
401 sut.scenario_helper.scenario_cfg = {
406 'allowed_drop_rate': '0.8 - 1',
410 'rules': 'acl_1rule.yaml',
414 'worker_config': '1C/1T',
419 'task_path': '/path/to/task'
422 @mock.patch.object(six.moves.builtins, 'open', create=True)
423 @mock.patch('yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.open',
424 mock.mock_open(), create=True)
425 @mock.patch('yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.LOG.exception')
426 def _traffic_runner(*args):
427 sut._setup_mq_producer = mock.Mock(return_value='mq_producer')
428 result = sut._traffic_runner(mock_traffic_profile, mock.ANY)
429 self.assertIsNone(result)
434 class TestIxiaBasicScenario(unittest.TestCase):
437 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
438 self.mock_IxNextgen = self._mock_IxNextgen.start()
439 self.context_cfg = mock.Mock()
440 self.ixia_cfg = mock.Mock()
441 self.scenario = tg_rfc2544_ixia.IxiaBasicScenario(self.mock_IxNextgen,
444 self.addCleanup(self._stop_mocks)
446 def _stop_mocks(self):
447 self._mock_IxNextgen.stop()
449 def test___init___(self):
450 self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaBasicScenario)
451 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
453 def test_apply_config(self):
454 self.assertIsNone(self.scenario.apply_config())
456 def test_create_traffic_model(self):
457 self.mock_IxNextgen.get_vports.return_value = [1, 2, 3, 4]
458 self.scenario.create_traffic_model()
459 self.scenario.client.get_vports.assert_called_once()
460 self.scenario.client.create_traffic_model.assert_called_once_with([1, 3], [2, 4])
462 def test_run_protocols(self):
463 self.assertIsNone(self.scenario.run_protocols())
465 def test_stop_protocols(self):
466 self.assertIsNone(self.scenario.stop_protocols())
469 class TestIxiaPppoeClientScenario(unittest.TestCase):
473 'sessions_per_port': 4,
474 'sessions_per_svlan': 1,
477 'ip': ['10.3.3.1', '10.4.4.1']
480 'sessions_per_port': 1,
481 'sessions_per_vlan': 1,
483 'gateway_ip': ['10.1.1.1', '10.2.2.1'],
484 'ip': ['10.1.1.1', '10.2.2.1'],
485 'prefix': ['24', '24']
491 'interfaces': {'xe0': {
492 'local_ip': '10.1.1.1',
493 'netmask': '255.255.255.0'
497 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
498 self.mock_IxNextgen = self._mock_IxNextgen.start()
499 self.scenario = tg_rfc2544_ixia.IxiaPppoeClientScenario(
500 self.mock_IxNextgen, self.CONTEXT_CFG, self.IXIA_CFG)
501 tg_rfc2544_ixia.WAIT_PROTOCOLS_STARTED = 2
502 self.addCleanup(self._stop_mocks)
504 def _stop_mocks(self):
505 self._mock_IxNextgen.stop()
507 def test___init___(self):
508 self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaPppoeClientScenario)
509 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
511 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
513 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
514 '_apply_access_network_config')
515 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
516 '_apply_core_network_config')
517 def test_apply_config(self, mock_apply_core_net_cfg,
518 mock_apply_access_net_cfg,
519 mock_fill_ixia_config):
520 self.mock_IxNextgen.get_vports.return_value = [1, 2, 3, 4]
521 self.scenario.apply_config()
522 self.scenario.client.get_vports.assert_called_once()
523 self.assertEqual(self.scenario._uplink_vports, [1, 3])
524 self.assertEqual(self.scenario._downlink_vports, [2, 4])
525 mock_fill_ixia_config.assert_called_once()
526 mock_apply_core_net_cfg.assert_called_once()
527 mock_apply_access_net_cfg.assert_called_once()
529 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
530 '_get_endpoints_src_dst_id_pairs')
531 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
532 '_get_endpoints_src_dst_obj_pairs')
533 def test_create_traffic_model(self, mock_obj_pairs, mock_id_pairs):
534 uplink_endpoints = ['group1', 'group2']
535 downlink_endpoints = ['group3', 'group3']
536 mock_id_pairs.return_value = ['xe0', 'xe1', 'xe0', 'xe1']
537 mock_obj_pairs.return_value = ['group1', 'group3', 'group2', 'group3']
538 mock_tp = mock.Mock()
539 mock_tp.full_profile = {'uplink_0': 'data',
540 'downlink_0': 'data',
544 self.scenario.create_traffic_model(mock_tp)
545 mock_id_pairs.assert_called_once_with(mock_tp.full_profile)
546 mock_obj_pairs.assert_called_once_with(['xe0', 'xe1', 'xe0', 'xe1'])
547 self.scenario.client.create_ipv4_traffic_model.assert_called_once_with(
548 uplink_endpoints, downlink_endpoints)
550 def test__get_endpoints_src_dst_id_pairs(self):
551 full_tp = OrderedDict([
552 ('uplink_0', {'ipv4': {'port': 'xe0'}}),
553 ('downlink_0', {'ipv4': {'port': 'xe1'}}),
554 ('uplink_1', {'ipv4': {'port': 'xe0'}}),
555 ('downlink_1', {'ipv4': {'port': 'xe3'}})])
556 endpoints_src_dst_pairs = ['xe0', 'xe1', 'xe0', 'xe3']
557 res = self.scenario._get_endpoints_src_dst_id_pairs(full_tp)
558 self.assertEqual(res, endpoints_src_dst_pairs)
560 def test__get_endpoints_src_dst_id_pairs_wrong_flows_number(self):
561 full_tp = OrderedDict([
562 ('uplink_0', {'ipv4': {'port': 'xe0'}}),
563 ('downlink_0', {'ipv4': {'port': 'xe1'}}),
564 ('uplink_1', {'ipv4': {'port': 'xe0'}})])
565 with self.assertRaises(RuntimeError):
566 self.scenario._get_endpoints_src_dst_id_pairs(full_tp)
568 def test__get_endpoints_src_dst_id_pairs_no_port_key(self):
569 full_tp = OrderedDict([
570 ('uplink_0', {'ipv4': {'id': 1}}),
571 ('downlink_0', {'ipv4': {'id': 2}})])
573 self.scenario._get_endpoints_src_dst_id_pairs(full_tp), [])
575 def test__get_endpoints_src_dst_obj_pairs_tp_with_port_key(self):
576 endpoints_id_pairs = ['xe0', 'xe1',
582 'sessions_per_port': 4,
583 'sessions_per_svlan': 1
586 'src_ip': [{'tg__0': 'xe0'}, {'tg__0': 'xe2'}],
587 'dst_ip': [{'tg__0': 'xe1'}, {'tg__0': 'xe3'}]
591 expected_result = ['tp1_dg1', 'tp3_dg1', 'tp1_dg2', 'tp3_dg1',
592 'tp1_dg3', 'tp4_dg1', 'tp1_dg4', 'tp4_dg1']
594 self.scenario._ixia_cfg = ixia_cfg
595 self.scenario._access_topologies = ['topology1', 'topology2']
596 self.scenario._core_topologies = ['topology3', 'topology4']
597 self.mock_IxNextgen.get_topology_device_groups.side_effect = \
598 [['tp1_dg1', 'tp1_dg2', 'tp1_dg3', 'tp1_dg4'],
599 ['tp2_dg1', 'tp2_dg2', 'tp2_dg3', 'tp2_dg4'],
602 res = self.scenario._get_endpoints_src_dst_obj_pairs(
604 self.assertEqual(res, expected_result)
606 def test__get_endpoints_src_dst_obj_pairs_default_flows_mapping(self):
607 endpoints_id_pairs = []
610 'sessions_per_port': 4,
611 'sessions_per_svlan': 1
614 'src_ip': [{'tg__0': 'xe0'}, {'tg__0': 'xe2'}],
615 'dst_ip': [{'tg__0': 'xe1'}, {'tg__0': 'xe3'}]
619 expected_result = ['tp1_dg1', 'tp3_dg1', 'tp1_dg2', 'tp3_dg1',
620 'tp1_dg3', 'tp3_dg1', 'tp1_dg4', 'tp3_dg1',
621 'tp2_dg1', 'tp4_dg1', 'tp2_dg2', 'tp4_dg1',
622 'tp2_dg3', 'tp4_dg1', 'tp2_dg4', 'tp4_dg1']
624 self.scenario._ixia_cfg = ixia_cfg
625 self.scenario._access_topologies = ['topology1', 'topology2']
626 self.scenario._core_topologies = ['topology3', 'topology4']
627 self.mock_IxNextgen.get_topology_device_groups.side_effect = \
628 [['tp1_dg1', 'tp1_dg2', 'tp1_dg3', 'tp1_dg4'],
629 ['tp2_dg1', 'tp2_dg2', 'tp2_dg3', 'tp2_dg4'],
632 res = self.scenario._get_endpoints_src_dst_obj_pairs(
634 self.assertEqual(res, expected_result)
636 def test_run_protocols(self):
637 self.scenario.client.is_protocols_running.return_value = True
638 self.scenario.run_protocols()
639 self.scenario.client.start_protocols.assert_called_once()
641 def test_run_protocols_timeout_exception(self):
642 self.scenario.client.is_protocols_running.return_value = False
643 with self.assertRaises(exceptions.WaitTimeout):
644 self.scenario.run_protocols()
645 self.scenario.client.start_protocols.assert_called_once()
647 def test_stop_protocols(self):
648 self.scenario.stop_protocols()
649 self.scenario.client.stop_protocols.assert_called_once()
651 def test__get_intf_addr_str_type_input(self):
652 intf = '192.168.10.2/24'
653 ip, mask = self.scenario._get_intf_addr(intf)
654 self.assertEqual(ip, '192.168.10.2')
655 self.assertEqual(mask, 24)
657 def test__get_intf_addr_dict_type_input(self):
658 intf = {'tg__0': 'xe0'}
659 ip, mask = self.scenario._get_intf_addr(intf)
660 self.assertEqual(ip, '10.1.1.1')
661 self.assertEqual(mask, 24)
663 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario, '_get_intf_addr')
664 def test__fill_ixia_config(self, mock_get_intf_addr):
668 'sessions_per_port': 4,
669 'sessions_per_svlan': 1,
672 'ip': ['10.3.3.1/24', '10.4.4.1/24']
675 'sessions_per_port': 1,
676 'sessions_per_vlan': 1,
678 'gateway_ip': ['10.1.1.1/24', '10.2.2.1/24'],
679 'ip': ['10.1.1.1/24', '10.2.2.1/24']
683 mock_get_intf_addr.side_effect = [
691 self.scenario._ixia_cfg = ixia_cfg
692 self.scenario._fill_ixia_config()
693 self.assertEqual(mock_get_intf_addr.call_count, 6)
694 self.assertEqual(self.scenario._ixia_cfg['pppoe_client']['ip'],
695 ['10.3.3.1', '10.4.4.1'])
696 self.assertEqual(self.scenario._ixia_cfg['ipv4_client']['ip'],
697 ['10.1.1.1', '10.2.2.1'])
698 self.assertEqual(self.scenario._ixia_cfg['ipv4_client']['prefix'],
701 @mock.patch('yardstick.network_services.libs.ixia_libs.ixnet.ixnet_api.Vlan')
702 def test__apply_access_network_config_pap_auth(self, mock_vlan):
705 'sessions_per_port': 4,
706 'sessions_per_svlan': 1,
709 'pap_user': 'test_pap',
710 'pap_password': 'pap'
712 pap_user = _ixia_cfg['pppoe_client']['pap_user']
713 pap_passwd = _ixia_cfg['pppoe_client']['pap_password']
714 self.scenario._ixia_cfg = _ixia_cfg
715 self.scenario._uplink_vports = [0, 2]
716 self.scenario.client.add_topology.side_effect = ['Topology 1', 'Topology 2']
717 self.scenario.client.add_device_group.side_effect = ['Dg1', 'Dg2', 'Dg3',
720 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2', 'Eth3',
721 'Eth4', 'Eth5', 'Eth6',
723 self.scenario._apply_access_network_config()
724 self.assertEqual(self.scenario.client.add_topology.call_count, 2)
725 self.assertEqual(self.scenario.client.add_device_group.call_count, 8)
726 self.assertEqual(self.scenario.client.add_ethernet.call_count, 8)
727 self.assertEqual(mock_vlan.call_count, 16)
728 self.assertEqual(self.scenario.client.add_vlans.call_count, 8)
729 self.assertEqual(self.scenario.client.add_pppox_client.call_count, 8)
730 self.scenario.client.add_topology.assert_has_calls([
731 mock.call('Topology access 0', 0),
732 mock.call('Topology access 1', 2)
734 self.scenario.client.add_device_group.assert_has_calls([
735 mock.call('Topology 1', 'SVLAN 10', 1),
736 mock.call('Topology 1', 'SVLAN 11', 1),
737 mock.call('Topology 1', 'SVLAN 12', 1),
738 mock.call('Topology 1', 'SVLAN 13', 1),
739 mock.call('Topology 2', 'SVLAN 14', 1),
740 mock.call('Topology 2', 'SVLAN 15', 1),
741 mock.call('Topology 2', 'SVLAN 16', 1),
742 mock.call('Topology 2', 'SVLAN 17', 1)
744 self.scenario.client.add_ethernet.assert_has_calls([
745 mock.call('Dg1', 'Ethernet'),
746 mock.call('Dg2', 'Ethernet'),
747 mock.call('Dg3', 'Ethernet'),
748 mock.call('Dg4', 'Ethernet'),
749 mock.call('Dg5', 'Ethernet'),
750 mock.call('Dg6', 'Ethernet'),
751 mock.call('Dg7', 'Ethernet'),
752 mock.call('Dg8', 'Ethernet')
754 mock_vlan.assert_has_calls([
755 mock.call(vlan_id=10),
756 mock.call(vlan_id=20, vlan_id_step=1),
757 mock.call(vlan_id=11),
758 mock.call(vlan_id=20, vlan_id_step=1),
759 mock.call(vlan_id=12),
760 mock.call(vlan_id=20, vlan_id_step=1),
761 mock.call(vlan_id=13),
762 mock.call(vlan_id=20, vlan_id_step=1),
763 mock.call(vlan_id=14),
764 mock.call(vlan_id=20, vlan_id_step=1),
765 mock.call(vlan_id=15),
766 mock.call(vlan_id=20, vlan_id_step=1),
767 mock.call(vlan_id=16),
768 mock.call(vlan_id=20, vlan_id_step=1),
769 mock.call(vlan_id=17),
770 mock.call(vlan_id=20, vlan_id_step=1)
772 self.scenario.client.add_pppox_client.assert_has_calls([
773 mock.call('Eth1', 'pap', pap_user, pap_passwd),
774 mock.call('Eth2', 'pap', pap_user, pap_passwd),
775 mock.call('Eth3', 'pap', pap_user, pap_passwd),
776 mock.call('Eth4', 'pap', pap_user, pap_passwd),
777 mock.call('Eth5', 'pap', pap_user, pap_passwd),
778 mock.call('Eth6', 'pap', pap_user, pap_passwd),
779 mock.call('Eth7', 'pap', pap_user, pap_passwd),
780 mock.call('Eth8', 'pap', pap_user, pap_passwd)
783 def test__apply_access_network_config_chap_auth(self):
786 'sessions_per_port': 4,
787 'sessions_per_svlan': 1,
790 'chap_user': 'test_chap',
791 'chap_password': 'chap'
793 chap_user = _ixia_cfg['pppoe_client']['chap_user']
794 chap_passwd = _ixia_cfg['pppoe_client']['chap_password']
795 self.scenario._ixia_cfg = _ixia_cfg
796 self.scenario._uplink_vports = [0, 2]
797 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2', 'Eth3',
798 'Eth4', 'Eth5', 'Eth6',
800 self.scenario._apply_access_network_config()
801 self.assertEqual(self.scenario.client.add_pppox_client.call_count, 8)
802 self.scenario.client.add_pppox_client.assert_has_calls([
803 mock.call('Eth1', 'chap', chap_user, chap_passwd),
804 mock.call('Eth2', 'chap', chap_user, chap_passwd),
805 mock.call('Eth3', 'chap', chap_user, chap_passwd),
806 mock.call('Eth4', 'chap', chap_user, chap_passwd),
807 mock.call('Eth5', 'chap', chap_user, chap_passwd),
808 mock.call('Eth6', 'chap', chap_user, chap_passwd),
809 mock.call('Eth7', 'chap', chap_user, chap_passwd),
810 mock.call('Eth8', 'chap', chap_user, chap_passwd)
813 @mock.patch('yardstick.network_services.libs.ixia_libs.ixnet.ixnet_api.Vlan')
814 def test__apply_core_network_config_no_bgp_proto(self, mock_vlan):
815 self.scenario._downlink_vports = [1, 3]
816 self.scenario.client.add_topology.side_effect = ['Topology 1', 'Topology 2']
817 self.scenario.client.add_device_group.side_effect = ['Dg1', 'Dg2']
818 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2']
819 self.scenario._apply_core_network_config()
820 self.assertEqual(self.scenario.client.add_topology.call_count, 2)
821 self.assertEqual(self.scenario.client.add_device_group.call_count, 2)
822 self.assertEqual(self.scenario.client.add_ethernet.call_count, 2)
823 self.assertEqual(mock_vlan.call_count, 2)
824 self.assertEqual(self.scenario.client.add_vlans.call_count, 2)
825 self.assertEqual(self.scenario.client.add_ipv4.call_count, 2)
826 self.scenario.client.add_topology.assert_has_calls([
827 mock.call('Topology core 0', 1),
828 mock.call('Topology core 1', 3)
830 self.scenario.client.add_device_group.assert_has_calls([
831 mock.call('Topology 1', 'Core port 0', 1),
832 mock.call('Topology 2', 'Core port 1', 1)
834 self.scenario.client.add_ethernet.assert_has_calls([
835 mock.call('Dg1', 'Ethernet'),
836 mock.call('Dg2', 'Ethernet')
838 mock_vlan.assert_has_calls([
839 mock.call(vlan_id=101),
840 mock.call(vlan_id=102)
842 self.scenario.client.add_ipv4.assert_has_calls([
843 mock.call('Eth1', name='ipv4', addr=ipaddress.IPv4Address('10.1.1.2'),
844 addr_step='0.0.0.1', prefix='24', gateway='10.1.1.1'),
845 mock.call('Eth2', name='ipv4', addr=ipaddress.IPv4Address('10.2.2.2'),
846 addr_step='0.0.0.1', prefix='24', gateway='10.2.2.1')
848 self.scenario.client.add_bgp.assert_not_called()
850 def test__apply_core_network_config_with_bgp_proto(self):
853 'bgp_type': 'external',
854 'dut_ip': '10.0.0.1',
858 self.scenario._ixia_cfg['ipv4_client'].update(bgp_params)
859 self.scenario._downlink_vports = [1, 3]
860 self.scenario.client.add_ipv4.side_effect = ['ipv4_1', 'ipv4_2']
861 self.scenario._apply_core_network_config()
862 self.assertEqual(self.scenario.client.add_bgp.call_count, 2)
863 self.scenario.client.add_bgp.assert_has_calls([
864 mock.call('ipv4_1', dut_ip=bgp_params["bgp"]["dut_ip"],
865 local_as=bgp_params["bgp"]["as_number"],
866 bgp_type=bgp_params["bgp"]["bgp_type"]),
867 mock.call('ipv4_2', dut_ip=bgp_params["bgp"]["dut_ip"],
868 local_as=bgp_params["bgp"]["as_number"],
869 bgp_type=bgp_params["bgp"]["bgp_type"])