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 STATS = {'stat_name': ['Card01/Port01',
497 'port_name': ['Ethernet - 001', 'Ethernet - 002'],
498 'Frames_Tx': ['150', '150'],
499 'Valid_Frames_Rx': ['150', '150'],
500 'Frames_Tx_Rate': ['0.0', '0.0'],
501 'Valid_Frames_Rx_Rate': ['0.0', '0.0'],
502 'Tx_Rate_Kbps': ['0.0', '0.0'],
503 'Rx_Rate_Mbps': ['0.0', '0.0'],
504 'Tx_Rate_Mbps': ['0.0', '0.0'],
505 'Rx_Rate_Kbps': ['0.0', '0.0'],
506 'Store-Forward_Max_latency_ns': ['100', '200'],
507 'Store-Forward_Min_latency_ns': ['100', '200'],
508 'Store-Forward_Avg_latency_ns': ['100', '200']}
511 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
512 self.mock_IxNextgen = self._mock_IxNextgen.start()
513 self.context_cfg = mock.Mock()
514 self.ixia_cfg = mock.Mock()
515 self.scenario = tg_rfc2544_ixia.IxiaBasicScenario(self.mock_IxNextgen,
518 self.addCleanup(self._stop_mocks)
520 def _stop_mocks(self):
521 self._mock_IxNextgen.stop()
523 def test___init___(self):
524 self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaBasicScenario)
525 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
527 def test_create_traffic_model(self):
528 self.mock_IxNextgen.get_vports.return_value = [1, 2, 3, 4]
529 self.scenario.create_traffic_model()
530 self.scenario.client.get_vports.assert_called_once()
531 self.scenario.client.create_traffic_model.assert_called_once_with([1, 3], [2, 4])
533 def test_apply_config(self):
534 self.assertIsNone(self.scenario.apply_config())
536 def test_run_protocols(self):
537 self.assertIsNone(self.scenario.run_protocols())
539 def test_stop_protocols(self):
540 self.assertIsNone(self.scenario.stop_protocols())
542 def test__get_stats(self):
543 self.scenario._get_stats()
544 self.scenario.client.get_statistics.assert_called_once()
546 @mock.patch.object(tg_rfc2544_ixia.IxiaBasicScenario, '_get_stats')
547 def test_generate_samples(self, mock_get_stats):
549 expected_samples = {'xe0': {
552 'rx_throughput_mbps': 0.0,
553 'rx_throughput_kps': 0.0,
556 'tx_throughput_mbps': 0.0,
557 'tx_throughput_kps': 0.0,
558 'Store-Forward_Max_latency_ns': 100,
559 'Store-Forward_Min_latency_ns': 100,
560 'Store-Forward_Avg_latency_ns': 100},
564 'rx_throughput_mbps': 0.0,
565 'rx_throughput_kps': 0.0,
568 'tx_throughput_mbps': 0.0,
569 'tx_throughput_kps': 0.0,
570 'Store-Forward_Max_latency_ns': 200,
571 'Store-Forward_Min_latency_ns': 200,
572 'Store-Forward_Avg_latency_ns': 200}}
574 res_helper = mock.Mock()
575 res_helper.vnfd_helper.find_interface_by_port.side_effect = \
576 [{'name': 'xe0'}, {'name': 'xe1'}]
579 mock_get_stats.return_value = self.STATS
580 samples = self.scenario.generate_samples(res_helper, ports, duration)
581 mock_get_stats.assert_called_once()
582 self.assertEqual(samples, expected_samples)
585 class TestIxiaL3Scenario(TestIxiaBasicScenario):
588 'src_ip': ['192.168.0.1-192.168.0.50'],
589 'dst_ip': ['192.168.1.1-192.168.1.150']
599 'vld_id': 'uplink_0',
600 'local_ip': '10.1.1.1',
601 'local_mac': 'aa:bb:cc:dd:ee:ff',
605 'vld_id': 'downlink_0',
606 'local_ip': '20.2.2.2',
607 'local_mac': 'bb:bb:cc:dd:ee:ee',
612 'network': "152.16.100.20",
613 'netmask': '255.255.0.0',
614 'gateway': '152.16.100.21',
622 super(TestIxiaL3Scenario, self).setUp()
623 self.ixia_cfg = self.IXIA_CFG
624 self.context_cfg = self.CONTEXT_CFG
625 self.scenario = tg_rfc2544_ixia.IxiaL3Scenario(self.mock_IxNextgen,
629 def test___init___(self):
630 self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaL3Scenario)
631 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
633 def test_create_traffic_model(self):
634 self.mock_IxNextgen.get_vports.return_value = ['1', '2']
635 self.scenario.create_traffic_model()
636 self.scenario.client.get_vports.assert_called_once()
637 self.scenario.client.create_ipv4_traffic_model.\
638 assert_called_once_with(['1/protocols/static'],
639 ['2/protocols/static'])
641 def test_apply_config(self):
642 self.scenario._add_interfaces = mock.Mock()
643 self.scenario._add_static_ips = mock.Mock()
644 self.assertIsNone(self.scenario.apply_config())
646 def test__add_static(self):
647 self.mock_IxNextgen.get_vports.return_value = ['1', '2']
648 self.mock_IxNextgen.get_static_interface.side_effect = ['intf1',
651 self.scenario._add_static_ips()
653 self.mock_IxNextgen.get_static_interface.assert_any_call('1')
654 self.mock_IxNextgen.get_static_interface.assert_any_call('2')
656 self.scenario.client.add_static_ipv4.assert_any_call(
657 'intf1', '1', '192.168.0.1', 49, '32')
658 self.scenario.client.add_static_ipv4.assert_any_call(
659 'intf2', '2', '192.168.1.1', 149, '32')
661 def test__add_interfaces(self):
662 self.mock_IxNextgen.get_vports.return_value = ['1', '2']
664 self.scenario._add_interfaces()
666 self.mock_IxNextgen.add_interface.assert_any_call('1',
670 self.mock_IxNextgen.add_interface.assert_any_call('2',
676 class TestIxiaPppoeClientScenario(unittest.TestCase):
680 'sessions_per_port': 4,
681 'sessions_per_svlan': 1,
684 'ip': ['10.3.3.1', '10.4.4.1']
687 'sessions_per_port': 1,
688 'sessions_per_vlan': 1,
690 'gateway_ip': ['10.1.1.1', '10.2.2.1'],
691 'ip': ['10.1.1.1', '10.2.2.1'],
692 'prefix': ['24', '24']
695 'tos': {'precedence': [0, 4]}
701 'interfaces': {'xe0': {
702 'local_ip': '10.1.1.1',
703 'netmask': '255.255.255.0'
707 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
708 self.mock_IxNextgen = self._mock_IxNextgen.start()
709 self.scenario = tg_rfc2544_ixia.IxiaPppoeClientScenario(
710 self.mock_IxNextgen, self.CONTEXT_CFG, self.IXIA_CFG)
711 tg_rfc2544_ixia.WAIT_PROTOCOLS_STARTED = 2
712 self.addCleanup(self._stop_mocks)
714 def _stop_mocks(self):
715 self._mock_IxNextgen.stop()
717 def test___init___(self):
718 self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaPppoeClientScenario)
719 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
721 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
723 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
724 '_apply_access_network_config')
725 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
726 '_apply_core_network_config')
727 def test_apply_config(self, mock_apply_core_net_cfg,
728 mock_apply_access_net_cfg,
729 mock_fill_ixia_config):
730 self.mock_IxNextgen.get_vports.return_value = [1, 2, 3, 4]
731 self.scenario.apply_config()
732 self.scenario.client.get_vports.assert_called_once()
733 self.assertEqual(self.scenario._uplink_vports, [1, 3])
734 self.assertEqual(self.scenario._downlink_vports, [2, 4])
735 mock_fill_ixia_config.assert_called_once()
736 mock_apply_core_net_cfg.assert_called_once()
737 mock_apply_access_net_cfg.assert_called_once()
739 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
740 '_get_endpoints_src_dst_id_pairs')
741 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
742 '_get_endpoints_src_dst_obj_pairs')
743 def test_create_traffic_model(self, mock_obj_pairs, mock_id_pairs):
744 uplink_endpoints = ['group1', 'group2']
745 downlink_endpoints = ['group3', 'group3']
746 mock_id_pairs.return_value = ['xe0', 'xe1', 'xe0', 'xe1']
747 mock_obj_pairs.return_value = ['group1', 'group3', 'group2', 'group3']
748 mock_tp = mock.Mock()
749 mock_tp.full_profile = {'uplink_0': 'data',
750 'downlink_0': 'data',
754 self.scenario.create_traffic_model(mock_tp)
755 mock_id_pairs.assert_called_once_with(mock_tp.full_profile)
756 mock_obj_pairs.assert_called_once_with(['xe0', 'xe1', 'xe0', 'xe1'])
757 self.scenario.client.create_ipv4_traffic_model.assert_called_once_with(
758 uplink_endpoints, downlink_endpoints)
760 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
761 '_get_endpoints_src_dst_id_pairs')
762 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
763 '_get_endpoints_src_dst_obj_pairs')
764 def test_create_traffic_model_topology_based_flows(self, mock_obj_pairs,
766 uplink_topologies = ['topology1', 'topology3']
767 downlink_topologies = ['topology2', 'topology4']
768 mock_id_pairs.return_value = []
769 mock_obj_pairs.return_value = []
770 mock_tp = mock.Mock()
771 mock_tp.full_profile = {'uplink_0': 'data',
772 'downlink_0': 'data',
776 self.scenario._access_topologies = ['topology1', 'topology3']
777 self.scenario._core_topologies = ['topology2', 'topology4']
778 self.scenario.create_traffic_model(mock_tp)
779 mock_id_pairs.assert_called_once_with(mock_tp.full_profile)
780 mock_obj_pairs.assert_called_once_with([])
781 self.scenario.client.create_ipv4_traffic_model.assert_called_once_with(
782 uplink_topologies, downlink_topologies)
784 def test__get_endpoints_src_dst_id_pairs(self):
785 full_tp = OrderedDict([
786 ('uplink_0', {'ipv4': {'port': 'xe0'}}),
787 ('downlink_0', {'ipv4': {'port': 'xe1'}}),
788 ('uplink_1', {'ipv4': {'port': 'xe0'}}),
789 ('downlink_1', {'ipv4': {'port': 'xe3'}})])
790 endpoints_src_dst_pairs = ['xe0', 'xe1', 'xe0', 'xe3']
791 res = self.scenario._get_endpoints_src_dst_id_pairs(full_tp)
792 self.assertEqual(res, endpoints_src_dst_pairs)
794 def test__get_endpoints_src_dst_id_pairs_wrong_flows_number(self):
795 full_tp = OrderedDict([
796 ('uplink_0', {'ipv4': {'port': 'xe0'}}),
797 ('downlink_0', {'ipv4': {'port': 'xe1'}}),
798 ('uplink_1', {'ipv4': {'port': 'xe0'}})])
799 with self.assertRaises(RuntimeError):
800 self.scenario._get_endpoints_src_dst_id_pairs(full_tp)
802 def test__get_endpoints_src_dst_id_pairs_no_port_key(self):
803 full_tp = OrderedDict([
804 ('uplink_0', {'ipv4': {'id': 1}}),
805 ('downlink_0', {'ipv4': {'id': 2}})])
807 self.scenario._get_endpoints_src_dst_id_pairs(full_tp), [])
809 def test__get_endpoints_src_dst_obj_pairs_tp_with_port_key(self):
810 endpoints_id_pairs = ['xe0', 'xe1',
816 'sessions_per_port': 4,
817 'sessions_per_svlan': 1
820 'src_ip': [{'tg__0': 'xe0'}, {'tg__0': 'xe2'}],
821 'dst_ip': [{'tg__0': 'xe1'}, {'tg__0': 'xe3'}]
825 expected_result = ['tp1_dg1', 'tp3_dg1', 'tp1_dg2', 'tp3_dg1',
826 'tp1_dg3', 'tp4_dg1', 'tp1_dg4', 'tp4_dg1']
828 self.scenario._ixia_cfg = ixia_cfg
829 self.scenario._access_topologies = ['topology1', 'topology2']
830 self.scenario._core_topologies = ['topology3', 'topology4']
831 self.mock_IxNextgen.get_topology_device_groups.side_effect = \
832 [['tp1_dg1', 'tp1_dg2', 'tp1_dg3', 'tp1_dg4'],
833 ['tp2_dg1', 'tp2_dg2', 'tp2_dg3', 'tp2_dg4'],
836 res = self.scenario._get_endpoints_src_dst_obj_pairs(
838 self.assertEqual(res, expected_result)
840 def test__get_endpoints_src_dst_obj_pairs_default_flows_mapping(self):
841 endpoints_id_pairs = []
844 'sessions_per_port': 4,
845 'sessions_per_svlan': 1
848 'src_ip': [{'tg__0': 'xe0'}, {'tg__0': 'xe2'}],
849 'dst_ip': [{'tg__0': 'xe1'}, {'tg__0': 'xe3'}]
853 self.scenario._ixia_cfg = ixia_cfg
854 res = self.scenario._get_endpoints_src_dst_obj_pairs(
856 self.assertEqual(res, [])
858 def test_run_protocols(self):
859 self.scenario.client.is_protocols_running.return_value = True
860 self.scenario.run_protocols()
861 self.scenario.client.start_protocols.assert_called_once()
863 def test_run_protocols_timeout_exception(self):
864 self.scenario.client.is_protocols_running.return_value = False
865 with self.assertRaises(exceptions.WaitTimeout):
866 self.scenario.run_protocols()
867 self.scenario.client.start_protocols.assert_called_once()
869 def test_stop_protocols(self):
870 self.scenario.stop_protocols()
871 self.scenario.client.stop_protocols.assert_called_once()
873 def test__get_intf_addr_str_type_input(self):
874 intf = '192.168.10.2/24'
875 ip, mask = self.scenario._get_intf_addr(intf)
876 self.assertEqual(ip, '192.168.10.2')
877 self.assertEqual(mask, 24)
879 def test__get_intf_addr_dict_type_input(self):
880 intf = {'tg__0': 'xe0'}
881 ip, mask = self.scenario._get_intf_addr(intf)
882 self.assertEqual(ip, '10.1.1.1')
883 self.assertEqual(mask, 24)
885 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario, '_get_intf_addr')
886 def test__fill_ixia_config(self, mock_get_intf_addr):
890 'sessions_per_port': 4,
891 'sessions_per_svlan': 1,
894 'ip': ['10.3.3.1/24', '10.4.4.1/24']
897 'sessions_per_port': 1,
898 'sessions_per_vlan': 1,
900 'gateway_ip': ['10.1.1.1/24', '10.2.2.1/24'],
901 'ip': ['10.1.1.1/24', '10.2.2.1/24']
905 mock_get_intf_addr.side_effect = [
913 self.scenario._ixia_cfg = ixia_cfg
914 self.scenario._fill_ixia_config()
915 self.assertEqual(mock_get_intf_addr.call_count, 6)
916 self.assertEqual(self.scenario._ixia_cfg['pppoe_client']['ip'],
917 ['10.3.3.1', '10.4.4.1'])
918 self.assertEqual(self.scenario._ixia_cfg['ipv4_client']['ip'],
919 ['10.1.1.1', '10.2.2.1'])
920 self.assertEqual(self.scenario._ixia_cfg['ipv4_client']['prefix'],
923 @mock.patch('yardstick.network_services.libs.ixia_libs.ixnet.ixnet_api.Vlan')
924 def test__apply_access_network_config_pap_auth(self, mock_vlan):
927 'sessions_per_port': 4,
928 'sessions_per_svlan': 1,
931 'pap_user': 'test_pap',
932 'pap_password': 'pap'
934 pap_user = _ixia_cfg['pppoe_client']['pap_user']
935 pap_passwd = _ixia_cfg['pppoe_client']['pap_password']
936 self.scenario._ixia_cfg = _ixia_cfg
937 self.scenario._uplink_vports = [0, 2]
938 self.scenario.client.add_topology.side_effect = ['Topology 1', 'Topology 2']
939 self.scenario.client.add_device_group.side_effect = ['Dg1', 'Dg2', 'Dg3',
942 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2', 'Eth3',
943 'Eth4', 'Eth5', 'Eth6',
945 self.scenario._apply_access_network_config()
946 self.assertEqual(self.scenario.client.add_topology.call_count, 2)
947 self.assertEqual(self.scenario.client.add_device_group.call_count, 8)
948 self.assertEqual(self.scenario.client.add_ethernet.call_count, 8)
949 self.assertEqual(mock_vlan.call_count, 16)
950 self.assertEqual(self.scenario.client.add_vlans.call_count, 8)
951 self.assertEqual(self.scenario.client.add_pppox_client.call_count, 8)
952 self.scenario.client.add_topology.assert_has_calls([
953 mock.call('Topology access 0', 0),
954 mock.call('Topology access 1', 2)
956 self.scenario.client.add_device_group.assert_has_calls([
957 mock.call('Topology 1', 'SVLAN 10', 1),
958 mock.call('Topology 1', 'SVLAN 11', 1),
959 mock.call('Topology 1', 'SVLAN 12', 1),
960 mock.call('Topology 1', 'SVLAN 13', 1),
961 mock.call('Topology 2', 'SVLAN 14', 1),
962 mock.call('Topology 2', 'SVLAN 15', 1),
963 mock.call('Topology 2', 'SVLAN 16', 1),
964 mock.call('Topology 2', 'SVLAN 17', 1)
966 self.scenario.client.add_ethernet.assert_has_calls([
967 mock.call('Dg1', 'Ethernet'),
968 mock.call('Dg2', 'Ethernet'),
969 mock.call('Dg3', 'Ethernet'),
970 mock.call('Dg4', 'Ethernet'),
971 mock.call('Dg5', 'Ethernet'),
972 mock.call('Dg6', 'Ethernet'),
973 mock.call('Dg7', 'Ethernet'),
974 mock.call('Dg8', 'Ethernet')
976 mock_vlan.assert_has_calls([
977 mock.call(vlan_id=10),
978 mock.call(vlan_id=20, vlan_id_step=1),
979 mock.call(vlan_id=11),
980 mock.call(vlan_id=20, vlan_id_step=1),
981 mock.call(vlan_id=12),
982 mock.call(vlan_id=20, vlan_id_step=1),
983 mock.call(vlan_id=13),
984 mock.call(vlan_id=20, vlan_id_step=1),
985 mock.call(vlan_id=14),
986 mock.call(vlan_id=20, vlan_id_step=1),
987 mock.call(vlan_id=15),
988 mock.call(vlan_id=20, vlan_id_step=1),
989 mock.call(vlan_id=16),
990 mock.call(vlan_id=20, vlan_id_step=1),
991 mock.call(vlan_id=17),
992 mock.call(vlan_id=20, vlan_id_step=1)
994 self.scenario.client.add_pppox_client.assert_has_calls([
995 mock.call('Eth1', 'pap', pap_user, pap_passwd),
996 mock.call('Eth2', 'pap', pap_user, pap_passwd),
997 mock.call('Eth3', 'pap', pap_user, pap_passwd),
998 mock.call('Eth4', 'pap', pap_user, pap_passwd),
999 mock.call('Eth5', 'pap', pap_user, pap_passwd),
1000 mock.call('Eth6', 'pap', pap_user, pap_passwd),
1001 mock.call('Eth7', 'pap', pap_user, pap_passwd),
1002 mock.call('Eth8', 'pap', pap_user, pap_passwd)
1005 def test__apply_access_network_config_chap_auth(self):
1008 'sessions_per_port': 4,
1009 'sessions_per_svlan': 1,
1012 'chap_user': 'test_chap',
1013 'chap_password': 'chap'
1015 chap_user = _ixia_cfg['pppoe_client']['chap_user']
1016 chap_passwd = _ixia_cfg['pppoe_client']['chap_password']
1017 self.scenario._ixia_cfg = _ixia_cfg
1018 self.scenario._uplink_vports = [0, 2]
1019 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2', 'Eth3',
1020 'Eth4', 'Eth5', 'Eth6',
1022 self.scenario._apply_access_network_config()
1023 self.assertEqual(self.scenario.client.add_pppox_client.call_count, 8)
1024 self.scenario.client.add_pppox_client.assert_has_calls([
1025 mock.call('Eth1', 'chap', chap_user, chap_passwd),
1026 mock.call('Eth2', 'chap', chap_user, chap_passwd),
1027 mock.call('Eth3', 'chap', chap_user, chap_passwd),
1028 mock.call('Eth4', 'chap', chap_user, chap_passwd),
1029 mock.call('Eth5', 'chap', chap_user, chap_passwd),
1030 mock.call('Eth6', 'chap', chap_user, chap_passwd),
1031 mock.call('Eth7', 'chap', chap_user, chap_passwd),
1032 mock.call('Eth8', 'chap', chap_user, chap_passwd)
1035 @mock.patch('yardstick.network_services.libs.ixia_libs.ixnet.ixnet_api.Vlan')
1036 def test__apply_core_network_config_no_bgp_proto(self, mock_vlan):
1037 self.scenario._downlink_vports = [1, 3]
1038 self.scenario.client.add_topology.side_effect = ['Topology 1', 'Topology 2']
1039 self.scenario.client.add_device_group.side_effect = ['Dg1', 'Dg2']
1040 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2']
1041 self.scenario._apply_core_network_config()
1042 self.assertEqual(self.scenario.client.add_topology.call_count, 2)
1043 self.assertEqual(self.scenario.client.add_device_group.call_count, 2)
1044 self.assertEqual(self.scenario.client.add_ethernet.call_count, 2)
1045 self.assertEqual(mock_vlan.call_count, 2)
1046 self.assertEqual(self.scenario.client.add_vlans.call_count, 2)
1047 self.assertEqual(self.scenario.client.add_ipv4.call_count, 2)
1048 self.scenario.client.add_topology.assert_has_calls([
1049 mock.call('Topology core 0', 1),
1050 mock.call('Topology core 1', 3)
1052 self.scenario.client.add_device_group.assert_has_calls([
1053 mock.call('Topology 1', 'Core port 0', 1),
1054 mock.call('Topology 2', 'Core port 1', 1)
1056 self.scenario.client.add_ethernet.assert_has_calls([
1057 mock.call('Dg1', 'Ethernet'),
1058 mock.call('Dg2', 'Ethernet')
1060 mock_vlan.assert_has_calls([
1061 mock.call(vlan_id=101),
1062 mock.call(vlan_id=102)
1064 self.scenario.client.add_ipv4.assert_has_calls([
1065 mock.call('Eth1', name='ipv4', addr=ipaddress.IPv4Address('10.1.1.2'),
1066 addr_step='0.0.0.1', prefix='24', gateway='10.1.1.1'),
1067 mock.call('Eth2', name='ipv4', addr=ipaddress.IPv4Address('10.2.2.2'),
1068 addr_step='0.0.0.1', prefix='24', gateway='10.2.2.1')
1070 self.scenario.client.add_bgp.assert_not_called()
1072 def test__apply_core_network_config_with_bgp_proto(self):
1075 'bgp_type': 'external',
1076 'dut_ip': '10.0.0.1',
1080 self.scenario._ixia_cfg['ipv4_client'].update(bgp_params)
1081 self.scenario._downlink_vports = [1, 3]
1082 self.scenario.client.add_ipv4.side_effect = ['ipv4_1', 'ipv4_2']
1083 self.scenario._apply_core_network_config()
1084 self.assertEqual(self.scenario.client.add_bgp.call_count, 2)
1085 self.scenario.client.add_bgp.assert_has_calls([
1086 mock.call('ipv4_1', dut_ip=bgp_params["bgp"]["dut_ip"],
1087 local_as=bgp_params["bgp"]["as_number"],
1088 bgp_type=bgp_params["bgp"]["bgp_type"]),
1089 mock.call('ipv4_2', dut_ip=bgp_params["bgp"]["dut_ip"],
1090 local_as=bgp_params["bgp"]["as_number"],
1091 bgp_type=bgp_params["bgp"]["bgp_type"])
1094 def test_update_tracking_options_raw_priority(self):
1095 raw_priority = {'raw': 4}
1096 self.scenario._ixia_cfg['priority'] = raw_priority
1097 self.scenario.update_tracking_options()
1098 self.scenario.client.set_flow_tracking.assert_called_once_with(
1099 ['flowGroup0', 'vlanVlanId0', 'ipv4Raw0'])
1101 def test_update_tracking_options_tos_priority(self):
1102 tos_priority = {'tos': {'precedence': [4, 7]}}
1103 self.scenario._ixia_cfg['priority'] = tos_priority
1104 self.scenario.update_tracking_options()
1105 self.scenario.client.set_flow_tracking.assert_called_once_with(
1106 ['flowGroup0', 'vlanVlanId0', 'ipv4Precedence0'])
1108 def test_update_tracking_options_dscp_priority(self):
1109 dscp_priority = {'dscp': {'defaultPHB': [4, 7]}}
1110 self.scenario._ixia_cfg['priority'] = dscp_priority
1111 self.scenario.update_tracking_options()
1112 self.scenario.client.set_flow_tracking.assert_called_once_with(
1113 ['flowGroup0', 'vlanVlanId0', 'ipv4DefaultPhb0'])
1115 def test_update_tracking_options_invalid_priority_data(self):
1116 invalid_priority = {'tos': {'inet-precedence': [4, 7]}}
1117 self.scenario._ixia_cfg['priority'] = invalid_priority
1118 self.scenario.update_tracking_options()
1119 self.scenario.client.set_flow_tracking.assert_called_once_with(
1120 ['flowGroup0', 'vlanVlanId0', 'ipv4Precedence0'])
1122 def test_get_tc_rfc2544_options(self):
1123 rfc2544_tc_opts = {'allowed_drop_rate': '0.0001 - 0.0001'}
1124 self.scenario._ixia_cfg['rfc2544'] = rfc2544_tc_opts
1125 res = self.scenario.get_tc_rfc2544_options()
1126 self.assertEqual(res, rfc2544_tc_opts)
1128 def test__get_stats(self):
1129 self.scenario._get_stats()
1130 self.scenario.client.get_pppoe_scenario_statistics.assert_called_once()
1132 def test_get_flow_id_data(self):
1133 stats = [{'id': 1, 'in_packets': 10, 'out_packets': 20}]
1136 res = self.scenario.get_flow_id_data(stats, flow_id, key)
1137 self.assertEqual(res, 10)
1139 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario, '_get_stats')
1140 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
1141 'get_priority_flows_stats')
1142 def test_generate_samples(self, mock_prio_flow_statistics,
1146 {'Flow_Group': 'RFC2544-1 - Flow Group 0001',
1147 'Frames_Delta': '0',
1149 'Rx_Frames': '3000',
1150 'Tx_Frames': '3000',
1152 'Tx_Port': 'Ethernet - 001',
1153 'Store-Forward_Avg_latency_ns': '2',
1154 'Store-Forward_Min_latency_ns': '2',
1155 'Store-Forward_Max_latency_ns': '2'},
1156 {'Flow_Group': 'RFC2544-2 - Flow Group 0001',
1157 'Frames_Delta': '0',
1159 'Rx_Frames': '3000',
1160 'Tx_Frames': '3000',
1162 'Tx_Port': 'Ethernet - 002',
1163 'Store-Forward_Avg_latency_ns': '2',
1164 'Store-Forward_Min_latency_ns': '2',
1165 'Store-Forward_Max_latency_ns': '2'
1167 'port_statistics': [
1168 {'Frames_Tx': '3000',
1169 'Valid_Frames_Rx': '3000',
1170 'Rx_Rate_Kbps': '0.0',
1171 'Tx_Rate_Kbps': '0.0',
1172 'Rx_Rate_Mbps': '0.0',
1173 'Tx_Rate_Mbps': '0.0',
1174 'port_name': 'Ethernet - 001'},
1175 {'Frames_Tx': '3000',
1176 'Valid_Frames_Rx': '3000',
1177 'Rx_Rate_Kbps': '0.0',
1178 'Tx_Rate_Kbps': '0.0',
1179 'Rx_Rate_Mbps': '0.0',
1180 'Tx_Rate_Mbps': '0.0',
1181 'port_name': 'Ethernet - 002'}],
1182 'pppox_client_per_port': [
1183 {'Sessions_Down': '0',
1184 'Sessions_Not_Started': '0',
1185 'Sessions_Total': '1',
1187 'subs_port': 'Ethernet - 001'}]}
1189 prio_flows_stats = {
1192 'out_packets': 6000,
1193 'RxThroughput': 200.0,
1194 'TxThroughput': 200.0,
1195 'avg_latency_ns': 2,
1196 'max_latency_ns': 2,
1201 expected_result = {'priority_stats': {
1202 '0': {'RxThroughput': 200.0,
1203 'TxThroughput': 200.0,
1204 'avg_latency_ns': 2,
1205 'max_latency_ns': 2,
1206 'min_latency_ns': 2,
1208 'out_packets': 6000}},
1209 'xe0': {'RxThroughput': 100.0,
1210 'Store-Forward_Avg_latency_ns': 2,
1211 'Store-Forward_Max_latency_ns': 2,
1212 'Store-Forward_Min_latency_ns': 2,
1213 'TxThroughput': 100.0,
1215 'out_packets': 3000,
1216 'rx_throughput_kps': 0.0,
1217 'rx_throughput_mbps': 0.0,
1219 'sessions_not_started': 0,
1220 'sessions_total': 1,
1222 'tx_throughput_kps': 0.0,
1223 'tx_throughput_mbps': 0.0},
1224 'xe1': {'RxThroughput': 100.0,
1225 'Store-Forward_Avg_latency_ns': 2,
1226 'Store-Forward_Max_latency_ns': 2,
1227 'Store-Forward_Min_latency_ns': 2,
1228 'TxThroughput': 100.0,
1230 'out_packets': 3000,
1231 'rx_throughput_kps': 0.0,
1232 'rx_throughput_mbps': 0.0,
1233 'tx_throughput_kps': 0.0,
1234 'tx_throughput_mbps': 0.0}}
1236 mock_get_stats.return_value = ixia_stats
1237 mock_prio_flow_statistics.return_value = prio_flows_stats
1239 port_names = [{'name': 'xe0'}, {'name': 'xe1'}]
1241 res_helper = mock.Mock()
1242 res_helper.vnfd_helper.find_interface_by_port.side_effect = \
1244 samples = self.scenario.generate_samples(res_helper, ports, duration)
1245 self.assertIsNotNone(samples)
1246 self.assertIsNotNone(samples.get('xe0'))
1247 self.assertIsNotNone(samples.get('xe1'))
1248 self.assertEqual(samples, expected_result)
1249 mock_get_stats.assert_called_once()
1250 mock_prio_flow_statistics.assert_called_once()