1 # Copyright (c) 2016-2019 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.
22 from collections import OrderedDict
24 from yardstick.common import utils
25 from yardstick.common import exceptions
26 from yardstick.benchmark import contexts
27 from yardstick.benchmark.contexts import base as ctx_base
28 from yardstick.network_services.libs.ixia_libs.ixnet import ixnet_api
29 from yardstick.network_services.traffic_profile import base as tp_base
30 from yardstick.network_services.vnf_generic.vnf import tg_rfc2544_ixia
33 TEST_FILE_YAML = 'nsb_test_case.yaml'
38 class TestIxiaResourceHelper(unittest.TestCase):
41 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
42 self.mock_IxNextgen = self._mock_IxNextgen.start()
43 self.addCleanup(self._stop_mocks)
45 def _stop_mocks(self):
46 self._mock_IxNextgen.stop()
48 def test___init___with_custom_rfc_helper(self):
49 class MyRfcHelper(tg_rfc2544_ixia.IxiaRfc2544Helper):
52 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(
53 mock.Mock(), MyRfcHelper)
54 self.assertIsInstance(ixia_resource_helper.rfc_helper, MyRfcHelper)
56 def test__init_ix_scenario(self):
57 mock_scenario = mock.Mock()
58 mock_scenario_helper = mock.Mock()
59 mock_scenario_helper.scenario_cfg = {'ixia_config': 'TestScenario',
60 'options': 'scenario_options'}
61 mock_setup_helper = mock.Mock(scenario_helper=mock_scenario_helper)
62 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock_setup_helper)
63 ixia_resource_helper._ixia_scenarios = {'TestScenario': mock_scenario}
64 ixia_resource_helper.client = 'client'
65 ixia_resource_helper.context_cfg = 'context'
66 ixia_resource_helper._init_ix_scenario()
67 mock_scenario.assert_called_once_with('client', 'context', 'scenario_options')
69 def test__init_ix_scenario_not_supported_cfg_type(self):
70 mock_scenario_helper = mock.Mock()
71 mock_scenario_helper.scenario_cfg = {'ixia_config': 'FakeScenario',
72 'options': 'scenario_options'}
73 mock_setup_helper = mock.Mock(scenario_helper=mock_scenario_helper)
74 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock_setup_helper)
75 ixia_resource_helper._ixia_scenarios = {'TestScenario': mock.Mock()}
76 with self.assertRaises(RuntimeError):
77 ixia_resource_helper._init_ix_scenario()
79 @mock.patch.object(tg_rfc2544_ixia.IxiaResourceHelper, '_init_ix_scenario')
80 def test_setup(self, mock__init_ix_scenario):
81 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
82 ixia_resource_helper.setup()
83 mock__init_ix_scenario.assert_called_once()
85 def test_stop_collect_with_client(self):
86 mock_client = mock.Mock()
87 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
88 ixia_resource_helper.client = mock_client
89 ixia_resource_helper._ix_scenario = mock.Mock()
90 ixia_resource_helper.stop_collect()
91 self.assertEqual(1, ixia_resource_helper._terminated.value)
92 ixia_resource_helper._ix_scenario.stop_protocols.assert_called_once()
94 def test_run_traffic(self):
95 mock_tprofile = mock.Mock()
96 mock_tprofile.config.duration = 10
97 mock_tprofile.get_drop_percentage.return_value = True, 'fake_samples'
98 ixia_rhelper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
99 ixia_rhelper.rfc_helper = mock.Mock()
100 ixia_rhelper.vnfd_helper = mock.Mock()
101 ixia_rhelper._ix_scenario = mock.Mock()
102 ixia_rhelper.vnfd_helper.port_pairs.all_ports = []
103 with mock.patch.object(ixia_rhelper, 'generate_samples'), \
104 mock.patch.object(ixia_rhelper, '_build_ports'), \
105 mock.patch.object(ixia_rhelper, '_initialize_client'), \
106 mock.patch.object(utils, 'wait_until_true'):
107 ixia_rhelper.run_traffic(mock_tprofile)
109 self.assertEqual('fake_samples', ixia_rhelper._queue.get())
110 mock_tprofile.update_traffic_profile.assert_called_once()
112 def test_run_test(self):
113 mock_tprofile = mock.Mock()
114 mock_tprofile.config.duration = 10
115 mock_tprofile.get_drop_percentage.return_value = True, 'fake_samples'
116 ixia_rhelper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
117 tasks_queue = mock.Mock()
118 tasks_queue.get.return_value = 'RUN_TRAFFIC'
119 results_queue = mock.Mock()
120 ixia_rhelper.rfc_helper = mock.Mock()
121 ixia_rhelper.vnfd_helper = mock.Mock()
122 ixia_rhelper._ix_scenario = mock.Mock()
123 ixia_rhelper.vnfd_helper.port_pairs.all_ports = []
124 with mock.patch.object(ixia_rhelper, 'generate_samples'), \
125 mock.patch.object(ixia_rhelper, '_build_ports'), \
126 mock.patch.object(ixia_rhelper, '_initialize_client'), \
127 mock.patch.object(utils, 'wait_until_true'):
128 ixia_rhelper.run_test(mock_tprofile, tasks_queue, results_queue)
130 self.assertEqual('fake_samples', ixia_rhelper._queue.get())
131 mock_tprofile.update_traffic_profile.assert_called_once()
132 tasks_queue.task_done.assert_called_once()
133 results_queue.put.assert_called_once_with('COMPLETE')
136 @mock.patch.object(tg_rfc2544_ixia, 'ixnet_api')
137 class TestIXIATrafficGen(unittest.TestCase):
138 VNFD = {'vnfd:vnfd-catalog':
140 [{'short-name': 'VpeVnf',
143 [{'network': '152.16.100.20',
144 'netmask': '255.255.255.0',
145 'gateway': '152.16.100.20',
147 {'network': '152.16.40.20',
148 'netmask': '255.255.255.0',
149 'gateway': '152.16.40.20',
151 'description': 'VPE approximation using DPDK',
152 'name': 'vpevnf-baremetal',
154 [{'network': '0064:ff9b:0:0:0:0:9810:6414',
156 'gateway': '0064:ff9b:0:0:0:0:9810:6414',
158 {'network': '0064:ff9b:0:0:0:0:9810:2814',
160 'gateway': '0064:ff9b:0:0:0:0:9810:2814',
162 'id': 'vpevnf-baremetal',
163 'external-interface':
164 [{'virtual-interface':
165 {'dst_mac': '00:00:00:00:00:04',
166 'vpci': '0000:05:00.0',
167 'local_ip': '152.16.100.19',
168 'type': 'PCI-PASSTHROUGH',
169 'netmask': '255.255.255.0',
171 'bandwidth': '10 Gbps',
173 'dst_ip': '152.16.100.20',
174 'local_iface_name': 'xe0',
175 'local_mac': '00:00:00:00:00:02'},
176 'vnfd-connection-point-ref': 'xe0',
178 {'virtual-interface':
179 {'dst_mac': '00:00:00:00:00:03',
180 'vpci': '0000:05:00.1',
181 'local_ip': '152.16.40.19',
182 'type': 'PCI-PASSTHROUGH',
184 'netmask': '255.255.255.0',
186 'bandwidth': '10 Gbps',
187 'dst_ip': '152.16.40.20',
188 'local_iface_name': 'xe1',
189 'local_mac': '00:00:00:00:00:01'},
190 'vnfd-connection-point-ref': 'xe1',
192 'description': 'Vpe approximation using DPDK',
194 {'vdu-id': 'vpevnf-baremetal',
200 {'kpi': ['packets_in', 'packets_fwd',
202 'connection-point': [{'type': 'VPORT', 'name': 'xe0'},
203 {'type': 'VPORT', 'name': 'xe1'}],
204 'id': 'VpeApproxVnf', 'name': 'VPEVnfSsh'}]}}
207 "schema": "isb:traffic_profile:0.1",
209 "description": "Fixed traffic profile to run UDP traffic",
211 "traffic_type": "FixedTraffic",
212 "frame_rate": 100, # pps
216 TC_YAML = {'scenarios': [{'tc_options':
217 {'rfc2544': {'allowed_drop_rate': '0.8 - 1'}},
218 'runner': {'duration': 400,
219 'interval': 35, 'type': 'Duration'},
221 {'flow': 'ipv4_1flow_Packets_vpe.yaml',
222 'imix': 'imix_voice.yaml'},
223 'vnf_options': {'vpe': {'cfg': 'vpe_config'}},
224 'traffic_profile': 'ipv4_throughput_vpe.yaml',
226 'nodes': {'tg__1': 'trafficgen_1.yardstick',
227 'vnf__1': 'vnf.yardstick'},
228 'topology': 'vpe_vnf_topology.yaml'}],
229 'context': {'nfvi_type': 'baremetal',
230 'type': contexts.CONTEXT_NODE,
232 'file': '/etc/yardstick/nodes/pod.yaml'},
233 'schema': 'yardstick:task:0.1'}
235 def test___init__(self, *args):
236 with mock.patch("yardstick.ssh.SSH") as ssh:
237 ssh_mock = mock.Mock(autospec=ssh.SSH)
239 mock.Mock(return_value=(0, "", ""))
240 ssh.from_node.return_value = ssh_mock
241 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
242 # NOTE(ralonsoh): check the object returned.
243 tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
245 def test_listen_traffic(self, *args):
246 with mock.patch("yardstick.ssh.SSH") as ssh:
247 ssh_mock = mock.Mock(autospec=ssh.SSH)
249 mock.Mock(return_value=(0, "", ""))
250 ssh.from_node.return_value = ssh_mock
251 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
252 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
253 self.assertIsNone(ixnet_traffic_gen.listen_traffic({}))
255 @mock.patch.object(ctx_base.Context, 'get_context_from_server', return_value='fake_context')
256 def test_instantiate(self, *args):
257 with mock.patch("yardstick.ssh.SSH") as ssh:
258 ssh_mock = mock.Mock(autospec=ssh.SSH)
260 mock.Mock(return_value=(0, "", ""))
262 mock.Mock(return_value=(0, "", ""))
263 ssh.from_node.return_value = ssh_mock
264 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
265 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
266 scenario_cfg = {'tc': "nsb_test_case",
274 'allowed_drop_rate': '0.8 - 1'},
276 'rules': 'acl_1rule.yaml',
280 'worker_config': '1C/1T',
281 'worker_threads': 1}}}})
282 scenario_cfg.update({
283 'nodes': {ixnet_traffic_gen.name: "mock"}
285 ixnet_traffic_gen.topology = ""
286 ixnet_traffic_gen.get_ixobj = mock.MagicMock()
287 ixnet_traffic_gen._ixia_traffic_gen = mock.MagicMock()
288 ixnet_traffic_gen._ixia_traffic_gen._connect = mock.Mock()
291 ixnet_traffic_gen.instantiate(scenario_cfg, {}))
293 @mock.patch.object(ctx_base.Context, 'get_physical_node_from_server', return_value='mock_node')
294 def test_collect_kpi(self, *args):
295 with mock.patch("yardstick.ssh.SSH") as ssh:
296 ssh_mock = mock.Mock(autospec=ssh.SSH)
298 mock.Mock(return_value=(0, "", ""))
299 ssh.from_node.return_value = ssh_mock
301 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
302 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
303 ixnet_traffic_gen.scenario_helper.scenario_cfg = {
304 'nodes': {ixnet_traffic_gen.name: "mock"}
306 ixnet_traffic_gen.data = {}
307 restult = ixnet_traffic_gen.collect_kpi()
309 expected = {'collect_stats': {},
310 'physical_node': 'mock_node'}
312 self.assertEqual(expected, restult)
314 def test_terminate(self, *args):
315 with mock.patch("yardstick.ssh.SSH") as ssh:
316 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
317 ssh_mock = mock.Mock(autospec=ssh.SSH)
319 mock.Mock(return_value=(0, "", ""))
320 ssh.from_node.return_value = ssh_mock
321 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(
322 NAME, vnfd, resource_helper_type=mock.Mock())
323 ixnet_traffic_gen._terminated = mock.MagicMock()
324 ixnet_traffic_gen._terminated.value = 0
325 ixnet_traffic_gen._ixia_traffic_gen = mock.MagicMock()
326 ixnet_traffic_gen._ixia_traffic_gen.ix_stop_traffic = mock.Mock()
327 ixnet_traffic_gen._traffic_process = mock.MagicMock()
328 ixnet_traffic_gen._traffic_process.terminate = mock.Mock()
329 self.assertIsNone(ixnet_traffic_gen.terminate())
331 def _get_file_abspath(self, filename):
332 curr_path = os.path.dirname(os.path.abspath(__file__))
333 file_path = os.path.join(curr_path, filename)
336 def test__check_status(self, *args):
337 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
338 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
341 @mock.patch("yardstick.ssh.SSH")
342 def test_traffic_runner(self, mock_ssh, *args):
343 mock_traffic_profile = mock.Mock(autospec=tp_base.TrafficProfile)
344 mock_traffic_profile.get_traffic_definition.return_value = "64"
345 mock_traffic_profile.params = self.TRAFFIC_PROFILE
346 # traffic_profile.ports is standardized on port_num
347 mock_traffic_profile.ports = [0, 1]
349 mock_ssh_instance = mock.Mock(autospec=mock_ssh.SSH)
350 mock_ssh_instance.execute.return_value = 0, "", ""
351 mock_ssh_instance.run.return_value = 0, "", ""
353 mock_ssh.from_node.return_value = mock_ssh_instance
355 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
356 vnfd["mgmt-interface"].update({
358 "ixchassis": "1.1.1.1",
359 "py_bin_path": "/root",
365 for ifname in range(1):
366 name = "xe{}".format(ifname)
374 "Valid_Frames_Rx": 1000,
380 samples.update({"CurrentDropPercentage": 0.0})
385 "Rx_Rate_Kbps": [20, 20],
386 "Tx_Rate_Kbps": [20, 20],
387 "Rx_Rate_Mbps": [10, 10],
388 "Tx_Rate_Mbps": [10, 10],
389 "CurrentDropPercentage": [0, 0],
390 "RxThroughput": [10, 10],
391 "TxThroughput": [10, 10],
392 "Frames_Tx": [1000, 1000],
393 "in_packets": [1000, 1000],
394 "Valid_Frames_Rx": [1000, 1000],
395 "out_packets": [1000, 1000],
399 mock_traffic_profile.execute_traffic.return_value = [
400 'Completed', samples]
401 mock_traffic_profile.get_drop_percentage.return_value = [
402 'Completed', samples]
404 sut = tg_rfc2544_ixia.IxiaTrafficGen(name, vnfd)
405 sut.vnf_port_pairs = [[[0], [1]]]
406 sut.tc_file_name = self._get_file_abspath(TEST_FILE_YAML)
409 sut.ssh_helper = mock.Mock()
410 sut._traffic_process = mock.MagicMock()
411 sut.generate_port_pairs = mock.Mock()
413 sut._ixia_traffic_gen = mock.MagicMock()
414 sut._ixia_traffic_gen.ix_get_statistics.return_value = last_res
416 sut.resource_helper.client = mock.MagicMock()
417 sut.resource_helper.client_started = mock.MagicMock()
418 sut.resource_helper.client_started.value = 1
419 sut.resource_helper.rfc_helper.iteration.value = 11
420 sut.resource_helper._ix_scenario = mock.Mock()
422 sut.scenario_helper.scenario_cfg = {
427 'allowed_drop_rate': '0.8 - 1',
431 'rules': 'acl_1rule.yaml',
435 'worker_config': '1C/1T',
440 'task_path': '/path/to/task'
443 @mock.patch.object(six.moves.builtins, 'open', create=True)
444 @mock.patch('yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.open',
445 mock.mock_open(), create=True)
446 @mock.patch('yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.LOG.exception')
447 def _traffic_runner(*args):
448 result = sut._traffic_runner(mock_traffic_profile)
449 self.assertIsNone(result)
453 def test_run_traffic_once(self, *args):
454 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
455 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
456 sut._init_traffic_process = mock.Mock()
457 sut._tasks_queue.put = mock.Mock()
458 sut.resource_helper.client_started.value = 0
459 sut.run_traffic_once(self.TRAFFIC_PROFILE)
460 sut._tasks_queue.put.assert_called_once_with("RUN_TRAFFIC")
461 sut._init_traffic_process.assert_called_once_with(self.TRAFFIC_PROFILE)
463 def test__test_runner(self, *args):
464 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
465 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
468 sut.resource_helper = mock.Mock()
469 sut._test_runner(self.TRAFFIC_PROFILE, tasks, results)
470 sut.resource_helper.run_test.assert_called_once_with(self.TRAFFIC_PROFILE,
473 @mock.patch.object(time, 'sleep', return_value=0)
474 def test__init_traffic_process(self, *args):
475 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
476 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
477 sut._test_runner = mock.Mock(return_value=0)
478 sut.resource_helper = mock.Mock()
479 sut.resource_helper.client_started.value = 0
480 sut._init_traffic_process(self.TRAFFIC_PROFILE)
482 def test_wait_on_traffic(self, *args):
483 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
484 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
485 sut._tasks_queue.join = mock.Mock(return_value=0)
486 sut._result_queue.get = mock.Mock(return_value='COMPLETE')
487 result = sut.wait_on_traffic()
488 sut._tasks_queue.join.assert_called_once()
489 sut._result_queue.get.assert_called_once()
490 self.assertEqual(result, 'COMPLETE')
493 class TestIxiaBasicScenario(unittest.TestCase):
495 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
496 self.mock_IxNextgen = self._mock_IxNextgen.start()
497 self.context_cfg = mock.Mock()
498 self.ixia_cfg = mock.Mock()
499 self.scenario = tg_rfc2544_ixia.IxiaBasicScenario(self.mock_IxNextgen,
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.IxiaBasicScenario)
509 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
511 def test_create_traffic_model(self):
512 self.mock_IxNextgen.get_vports.return_value = [1, 2, 3, 4]
513 self.scenario.create_traffic_model()
514 self.scenario.client.get_vports.assert_called_once()
515 self.scenario.client.create_traffic_model.assert_called_once_with([1, 3], [2, 4])
517 def test_apply_config(self):
518 self.assertIsNone(self.scenario.apply_config())
520 def test_run_protocols(self):
521 self.assertIsNone(self.scenario.run_protocols())
523 def test_stop_protocols(self):
524 self.assertIsNone(self.scenario.stop_protocols())
527 class TestIxiaL3Scenario(TestIxiaBasicScenario):
530 'src_ip': ['192.168.0.1-192.168.0.50'],
531 'dst_ip': ['192.168.1.1-192.168.1.150']
541 'vld_id': 'uplink_0',
542 'local_ip': '10.1.1.1',
543 'local_mac': 'aa:bb:cc:dd:ee:ff',
547 'vld_id': 'downlink_0',
548 'local_ip': '20.2.2.2',
549 'local_mac': 'bb:bb:cc:dd:ee:ee',
554 'network': "152.16.100.20",
555 'netmask': '255.255.0.0',
556 'gateway': '152.16.100.21',
564 super(TestIxiaL3Scenario, self).setUp()
565 self.ixia_cfg = self.IXIA_CFG
566 self.context_cfg = self.CONTEXT_CFG
567 self.scenario = tg_rfc2544_ixia.IxiaL3Scenario(self.mock_IxNextgen,
571 def test___init___(self):
572 self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaL3Scenario)
573 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
575 def test_create_traffic_model(self):
576 self.mock_IxNextgen.get_vports.return_value = ['1', '2']
577 self.scenario.create_traffic_model()
578 self.scenario.client.get_vports.assert_called_once()
579 self.scenario.client.create_ipv4_traffic_model.\
580 assert_called_once_with(['1/protocols/static'],
581 ['2/protocols/static'])
583 def test_apply_config(self):
584 self.scenario._add_interfaces = mock.Mock()
585 self.scenario._add_static_ips = mock.Mock()
586 self.assertIsNone(self.scenario.apply_config())
588 def test__add_static(self):
589 self.mock_IxNextgen.get_vports.return_value = ['1', '2']
590 self.mock_IxNextgen.get_static_interface.side_effect = ['intf1',
593 self.scenario._add_static_ips()
595 self.mock_IxNextgen.get_static_interface.assert_any_call('1')
596 self.mock_IxNextgen.get_static_interface.assert_any_call('2')
598 self.scenario.client.add_static_ipv4.assert_any_call(
599 'intf1', '1', '192.168.0.1', 49, '32')
600 self.scenario.client.add_static_ipv4.assert_any_call(
601 'intf2', '2', '192.168.1.1', 149, '32')
603 def test__add_interfaces(self):
604 self.mock_IxNextgen.get_vports.return_value = ['1', '2']
606 self.scenario._add_interfaces()
608 self.mock_IxNextgen.add_interface.assert_any_call('1',
612 self.mock_IxNextgen.add_interface.assert_any_call('2',
618 class TestIxiaPppoeClientScenario(unittest.TestCase):
622 'sessions_per_port': 4,
623 'sessions_per_svlan': 1,
626 'ip': ['10.3.3.1', '10.4.4.1']
629 'sessions_per_port': 1,
630 'sessions_per_vlan': 1,
632 'gateway_ip': ['10.1.1.1', '10.2.2.1'],
633 'ip': ['10.1.1.1', '10.2.2.1'],
634 'prefix': ['24', '24']
640 'interfaces': {'xe0': {
641 'local_ip': '10.1.1.1',
642 'netmask': '255.255.255.0'
646 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
647 self.mock_IxNextgen = self._mock_IxNextgen.start()
648 self.scenario = tg_rfc2544_ixia.IxiaPppoeClientScenario(
649 self.mock_IxNextgen, self.CONTEXT_CFG, self.IXIA_CFG)
650 tg_rfc2544_ixia.WAIT_PROTOCOLS_STARTED = 2
651 self.addCleanup(self._stop_mocks)
653 def _stop_mocks(self):
654 self._mock_IxNextgen.stop()
656 def test___init___(self):
657 self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaPppoeClientScenario)
658 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
660 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
662 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
663 '_apply_access_network_config')
664 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
665 '_apply_core_network_config')
666 def test_apply_config(self, mock_apply_core_net_cfg,
667 mock_apply_access_net_cfg,
668 mock_fill_ixia_config):
669 self.mock_IxNextgen.get_vports.return_value = [1, 2, 3, 4]
670 self.scenario.apply_config()
671 self.scenario.client.get_vports.assert_called_once()
672 self.assertEqual(self.scenario._uplink_vports, [1, 3])
673 self.assertEqual(self.scenario._downlink_vports, [2, 4])
674 mock_fill_ixia_config.assert_called_once()
675 mock_apply_core_net_cfg.assert_called_once()
676 mock_apply_access_net_cfg.assert_called_once()
678 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
679 '_get_endpoints_src_dst_id_pairs')
680 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
681 '_get_endpoints_src_dst_obj_pairs')
682 def test_create_traffic_model(self, mock_obj_pairs, mock_id_pairs):
683 uplink_endpoints = ['group1', 'group2']
684 downlink_endpoints = ['group3', 'group3']
685 mock_id_pairs.return_value = ['xe0', 'xe1', 'xe0', 'xe1']
686 mock_obj_pairs.return_value = ['group1', 'group3', 'group2', 'group3']
687 mock_tp = mock.Mock()
688 mock_tp.full_profile = {'uplink_0': 'data',
689 'downlink_0': 'data',
693 self.scenario.create_traffic_model(mock_tp)
694 mock_id_pairs.assert_called_once_with(mock_tp.full_profile)
695 mock_obj_pairs.assert_called_once_with(['xe0', 'xe1', 'xe0', 'xe1'])
696 self.scenario.client.create_ipv4_traffic_model.assert_called_once_with(
697 uplink_endpoints, downlink_endpoints)
699 def test__get_endpoints_src_dst_id_pairs(self):
700 full_tp = OrderedDict([
701 ('uplink_0', {'ipv4': {'port': 'xe0'}}),
702 ('downlink_0', {'ipv4': {'port': 'xe1'}}),
703 ('uplink_1', {'ipv4': {'port': 'xe0'}}),
704 ('downlink_1', {'ipv4': {'port': 'xe3'}})])
705 endpoints_src_dst_pairs = ['xe0', 'xe1', 'xe0', 'xe3']
706 res = self.scenario._get_endpoints_src_dst_id_pairs(full_tp)
707 self.assertEqual(res, endpoints_src_dst_pairs)
709 def test__get_endpoints_src_dst_id_pairs_wrong_flows_number(self):
710 full_tp = OrderedDict([
711 ('uplink_0', {'ipv4': {'port': 'xe0'}}),
712 ('downlink_0', {'ipv4': {'port': 'xe1'}}),
713 ('uplink_1', {'ipv4': {'port': 'xe0'}})])
714 with self.assertRaises(RuntimeError):
715 self.scenario._get_endpoints_src_dst_id_pairs(full_tp)
717 def test__get_endpoints_src_dst_id_pairs_no_port_key(self):
718 full_tp = OrderedDict([
719 ('uplink_0', {'ipv4': {'id': 1}}),
720 ('downlink_0', {'ipv4': {'id': 2}})])
722 self.scenario._get_endpoints_src_dst_id_pairs(full_tp), [])
724 def test__get_endpoints_src_dst_obj_pairs_tp_with_port_key(self):
725 endpoints_id_pairs = ['xe0', 'xe1',
731 'sessions_per_port': 4,
732 'sessions_per_svlan': 1
735 'src_ip': [{'tg__0': 'xe0'}, {'tg__0': 'xe2'}],
736 'dst_ip': [{'tg__0': 'xe1'}, {'tg__0': 'xe3'}]
740 expected_result = ['tp1_dg1', 'tp3_dg1', 'tp1_dg2', 'tp3_dg1',
741 'tp1_dg3', 'tp4_dg1', 'tp1_dg4', 'tp4_dg1']
743 self.scenario._ixia_cfg = ixia_cfg
744 self.scenario._access_topologies = ['topology1', 'topology2']
745 self.scenario._core_topologies = ['topology3', 'topology4']
746 self.mock_IxNextgen.get_topology_device_groups.side_effect = \
747 [['tp1_dg1', 'tp1_dg2', 'tp1_dg3', 'tp1_dg4'],
748 ['tp2_dg1', 'tp2_dg2', 'tp2_dg3', 'tp2_dg4'],
751 res = self.scenario._get_endpoints_src_dst_obj_pairs(
753 self.assertEqual(res, expected_result)
755 def test__get_endpoints_src_dst_obj_pairs_default_flows_mapping(self):
756 endpoints_id_pairs = []
759 'sessions_per_port': 4,
760 'sessions_per_svlan': 1
763 'src_ip': [{'tg__0': 'xe0'}, {'tg__0': 'xe2'}],
764 'dst_ip': [{'tg__0': 'xe1'}, {'tg__0': 'xe3'}]
768 expected_result = ['tp1_dg1', 'tp3_dg1', 'tp1_dg2', 'tp3_dg1',
769 'tp1_dg3', 'tp3_dg1', 'tp1_dg4', 'tp3_dg1',
770 'tp2_dg1', 'tp4_dg1', 'tp2_dg2', 'tp4_dg1',
771 'tp2_dg3', 'tp4_dg1', 'tp2_dg4', 'tp4_dg1']
773 self.scenario._ixia_cfg = ixia_cfg
774 self.scenario._access_topologies = ['topology1', 'topology2']
775 self.scenario._core_topologies = ['topology3', 'topology4']
776 self.mock_IxNextgen.get_topology_device_groups.side_effect = \
777 [['tp1_dg1', 'tp1_dg2', 'tp1_dg3', 'tp1_dg4'],
778 ['tp2_dg1', 'tp2_dg2', 'tp2_dg3', 'tp2_dg4'],
781 res = self.scenario._get_endpoints_src_dst_obj_pairs(
783 self.assertEqual(res, expected_result)
785 def test_run_protocols(self):
786 self.scenario.client.is_protocols_running.return_value = True
787 self.scenario.run_protocols()
788 self.scenario.client.start_protocols.assert_called_once()
790 def test_run_protocols_timeout_exception(self):
791 self.scenario.client.is_protocols_running.return_value = False
792 with self.assertRaises(exceptions.WaitTimeout):
793 self.scenario.run_protocols()
794 self.scenario.client.start_protocols.assert_called_once()
796 def test_stop_protocols(self):
797 self.scenario.stop_protocols()
798 self.scenario.client.stop_protocols.assert_called_once()
800 def test__get_intf_addr_str_type_input(self):
801 intf = '192.168.10.2/24'
802 ip, mask = self.scenario._get_intf_addr(intf)
803 self.assertEqual(ip, '192.168.10.2')
804 self.assertEqual(mask, 24)
806 def test__get_intf_addr_dict_type_input(self):
807 intf = {'tg__0': 'xe0'}
808 ip, mask = self.scenario._get_intf_addr(intf)
809 self.assertEqual(ip, '10.1.1.1')
810 self.assertEqual(mask, 24)
812 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario, '_get_intf_addr')
813 def test__fill_ixia_config(self, mock_get_intf_addr):
817 'sessions_per_port': 4,
818 'sessions_per_svlan': 1,
821 'ip': ['10.3.3.1/24', '10.4.4.1/24']
824 'sessions_per_port': 1,
825 'sessions_per_vlan': 1,
827 'gateway_ip': ['10.1.1.1/24', '10.2.2.1/24'],
828 'ip': ['10.1.1.1/24', '10.2.2.1/24']
832 mock_get_intf_addr.side_effect = [
840 self.scenario._ixia_cfg = ixia_cfg
841 self.scenario._fill_ixia_config()
842 self.assertEqual(mock_get_intf_addr.call_count, 6)
843 self.assertEqual(self.scenario._ixia_cfg['pppoe_client']['ip'],
844 ['10.3.3.1', '10.4.4.1'])
845 self.assertEqual(self.scenario._ixia_cfg['ipv4_client']['ip'],
846 ['10.1.1.1', '10.2.2.1'])
847 self.assertEqual(self.scenario._ixia_cfg['ipv4_client']['prefix'],
850 @mock.patch('yardstick.network_services.libs.ixia_libs.ixnet.ixnet_api.Vlan')
851 def test__apply_access_network_config_pap_auth(self, mock_vlan):
854 'sessions_per_port': 4,
855 'sessions_per_svlan': 1,
858 'pap_user': 'test_pap',
859 'pap_password': 'pap'
861 pap_user = _ixia_cfg['pppoe_client']['pap_user']
862 pap_passwd = _ixia_cfg['pppoe_client']['pap_password']
863 self.scenario._ixia_cfg = _ixia_cfg
864 self.scenario._uplink_vports = [0, 2]
865 self.scenario.client.add_topology.side_effect = ['Topology 1', 'Topology 2']
866 self.scenario.client.add_device_group.side_effect = ['Dg1', 'Dg2', 'Dg3',
869 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2', 'Eth3',
870 'Eth4', 'Eth5', 'Eth6',
872 self.scenario._apply_access_network_config()
873 self.assertEqual(self.scenario.client.add_topology.call_count, 2)
874 self.assertEqual(self.scenario.client.add_device_group.call_count, 8)
875 self.assertEqual(self.scenario.client.add_ethernet.call_count, 8)
876 self.assertEqual(mock_vlan.call_count, 16)
877 self.assertEqual(self.scenario.client.add_vlans.call_count, 8)
878 self.assertEqual(self.scenario.client.add_pppox_client.call_count, 8)
879 self.scenario.client.add_topology.assert_has_calls([
880 mock.call('Topology access 0', 0),
881 mock.call('Topology access 1', 2)
883 self.scenario.client.add_device_group.assert_has_calls([
884 mock.call('Topology 1', 'SVLAN 10', 1),
885 mock.call('Topology 1', 'SVLAN 11', 1),
886 mock.call('Topology 1', 'SVLAN 12', 1),
887 mock.call('Topology 1', 'SVLAN 13', 1),
888 mock.call('Topology 2', 'SVLAN 14', 1),
889 mock.call('Topology 2', 'SVLAN 15', 1),
890 mock.call('Topology 2', 'SVLAN 16', 1),
891 mock.call('Topology 2', 'SVLAN 17', 1)
893 self.scenario.client.add_ethernet.assert_has_calls([
894 mock.call('Dg1', 'Ethernet'),
895 mock.call('Dg2', 'Ethernet'),
896 mock.call('Dg3', 'Ethernet'),
897 mock.call('Dg4', 'Ethernet'),
898 mock.call('Dg5', 'Ethernet'),
899 mock.call('Dg6', 'Ethernet'),
900 mock.call('Dg7', 'Ethernet'),
901 mock.call('Dg8', 'Ethernet')
903 mock_vlan.assert_has_calls([
904 mock.call(vlan_id=10),
905 mock.call(vlan_id=20, vlan_id_step=1),
906 mock.call(vlan_id=11),
907 mock.call(vlan_id=20, vlan_id_step=1),
908 mock.call(vlan_id=12),
909 mock.call(vlan_id=20, vlan_id_step=1),
910 mock.call(vlan_id=13),
911 mock.call(vlan_id=20, vlan_id_step=1),
912 mock.call(vlan_id=14),
913 mock.call(vlan_id=20, vlan_id_step=1),
914 mock.call(vlan_id=15),
915 mock.call(vlan_id=20, vlan_id_step=1),
916 mock.call(vlan_id=16),
917 mock.call(vlan_id=20, vlan_id_step=1),
918 mock.call(vlan_id=17),
919 mock.call(vlan_id=20, vlan_id_step=1)
921 self.scenario.client.add_pppox_client.assert_has_calls([
922 mock.call('Eth1', 'pap', pap_user, pap_passwd),
923 mock.call('Eth2', 'pap', pap_user, pap_passwd),
924 mock.call('Eth3', 'pap', pap_user, pap_passwd),
925 mock.call('Eth4', 'pap', pap_user, pap_passwd),
926 mock.call('Eth5', 'pap', pap_user, pap_passwd),
927 mock.call('Eth6', 'pap', pap_user, pap_passwd),
928 mock.call('Eth7', 'pap', pap_user, pap_passwd),
929 mock.call('Eth8', 'pap', pap_user, pap_passwd)
932 def test__apply_access_network_config_chap_auth(self):
935 'sessions_per_port': 4,
936 'sessions_per_svlan': 1,
939 'chap_user': 'test_chap',
940 'chap_password': 'chap'
942 chap_user = _ixia_cfg['pppoe_client']['chap_user']
943 chap_passwd = _ixia_cfg['pppoe_client']['chap_password']
944 self.scenario._ixia_cfg = _ixia_cfg
945 self.scenario._uplink_vports = [0, 2]
946 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2', 'Eth3',
947 'Eth4', 'Eth5', 'Eth6',
949 self.scenario._apply_access_network_config()
950 self.assertEqual(self.scenario.client.add_pppox_client.call_count, 8)
951 self.scenario.client.add_pppox_client.assert_has_calls([
952 mock.call('Eth1', 'chap', chap_user, chap_passwd),
953 mock.call('Eth2', 'chap', chap_user, chap_passwd),
954 mock.call('Eth3', 'chap', chap_user, chap_passwd),
955 mock.call('Eth4', 'chap', chap_user, chap_passwd),
956 mock.call('Eth5', 'chap', chap_user, chap_passwd),
957 mock.call('Eth6', 'chap', chap_user, chap_passwd),
958 mock.call('Eth7', 'chap', chap_user, chap_passwd),
959 mock.call('Eth8', 'chap', chap_user, chap_passwd)
962 @mock.patch('yardstick.network_services.libs.ixia_libs.ixnet.ixnet_api.Vlan')
963 def test__apply_core_network_config_no_bgp_proto(self, mock_vlan):
964 self.scenario._downlink_vports = [1, 3]
965 self.scenario.client.add_topology.side_effect = ['Topology 1', 'Topology 2']
966 self.scenario.client.add_device_group.side_effect = ['Dg1', 'Dg2']
967 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2']
968 self.scenario._apply_core_network_config()
969 self.assertEqual(self.scenario.client.add_topology.call_count, 2)
970 self.assertEqual(self.scenario.client.add_device_group.call_count, 2)
971 self.assertEqual(self.scenario.client.add_ethernet.call_count, 2)
972 self.assertEqual(mock_vlan.call_count, 2)
973 self.assertEqual(self.scenario.client.add_vlans.call_count, 2)
974 self.assertEqual(self.scenario.client.add_ipv4.call_count, 2)
975 self.scenario.client.add_topology.assert_has_calls([
976 mock.call('Topology core 0', 1),
977 mock.call('Topology core 1', 3)
979 self.scenario.client.add_device_group.assert_has_calls([
980 mock.call('Topology 1', 'Core port 0', 1),
981 mock.call('Topology 2', 'Core port 1', 1)
983 self.scenario.client.add_ethernet.assert_has_calls([
984 mock.call('Dg1', 'Ethernet'),
985 mock.call('Dg2', 'Ethernet')
987 mock_vlan.assert_has_calls([
988 mock.call(vlan_id=101),
989 mock.call(vlan_id=102)
991 self.scenario.client.add_ipv4.assert_has_calls([
992 mock.call('Eth1', name='ipv4', addr=ipaddress.IPv4Address('10.1.1.2'),
993 addr_step='0.0.0.1', prefix='24', gateway='10.1.1.1'),
994 mock.call('Eth2', name='ipv4', addr=ipaddress.IPv4Address('10.2.2.2'),
995 addr_step='0.0.0.1', prefix='24', gateway='10.2.2.1')
997 self.scenario.client.add_bgp.assert_not_called()
999 def test__apply_core_network_config_with_bgp_proto(self):
1002 'bgp_type': 'external',
1003 'dut_ip': '10.0.0.1',
1007 self.scenario._ixia_cfg['ipv4_client'].update(bgp_params)
1008 self.scenario._downlink_vports = [1, 3]
1009 self.scenario.client.add_ipv4.side_effect = ['ipv4_1', 'ipv4_2']
1010 self.scenario._apply_core_network_config()
1011 self.assertEqual(self.scenario.client.add_bgp.call_count, 2)
1012 self.scenario.client.add_bgp.assert_has_calls([
1013 mock.call('ipv4_1', dut_ip=bgp_params["bgp"]["dut_ip"],
1014 local_as=bgp_params["bgp"]["as_number"],
1015 bgp_type=bgp_params["bgp"]["bgp_type"]),
1016 mock.call('ipv4_2', dut_ip=bgp_params["bgp"]["dut_ip"],
1017 local_as=bgp_params["bgp"]["as_number"],
1018 bgp_type=bgp_params["bgp"]["bgp_type"])