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
31 from yardstick.network_services.traffic_profile import ixia_rfc2544
34 TEST_FILE_YAML = 'nsb_test_case.yaml'
39 class TestIxiaResourceHelper(unittest.TestCase):
42 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
43 self.mock_IxNextgen = self._mock_IxNextgen.start()
44 self.addCleanup(self._stop_mocks)
46 def _stop_mocks(self):
47 self._mock_IxNextgen.stop()
49 def test___init___with_custom_rfc_helper(self):
50 class MyRfcHelper(tg_rfc2544_ixia.IxiaRfc2544Helper):
53 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(
54 mock.Mock(), MyRfcHelper)
55 self.assertIsInstance(ixia_resource_helper.rfc_helper, MyRfcHelper)
57 def test__init_ix_scenario(self):
58 mock_scenario = mock.Mock()
59 mock_scenario_helper = mock.Mock()
60 mock_scenario_helper.scenario_cfg = {'ixia_config': 'TestScenario',
61 'options': 'scenario_options'}
62 mock_setup_helper = mock.Mock(scenario_helper=mock_scenario_helper)
63 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock_setup_helper)
64 ixia_resource_helper._ixia_scenarios = {'TestScenario': mock_scenario}
65 ixia_resource_helper.client = 'client'
66 ixia_resource_helper.context_cfg = 'context'
67 ixia_resource_helper._init_ix_scenario()
68 mock_scenario.assert_called_once_with('client', 'context', 'scenario_options')
70 def test__init_ix_scenario_not_supported_cfg_type(self):
71 mock_scenario_helper = mock.Mock()
72 mock_scenario_helper.scenario_cfg = {'ixia_config': 'FakeScenario',
73 'options': 'scenario_options'}
74 mock_setup_helper = mock.Mock(scenario_helper=mock_scenario_helper)
75 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock_setup_helper)
76 ixia_resource_helper._ixia_scenarios = {'TestScenario': mock.Mock()}
77 with self.assertRaises(RuntimeError):
78 ixia_resource_helper._init_ix_scenario()
80 @mock.patch.object(tg_rfc2544_ixia.IxiaResourceHelper, '_init_ix_scenario')
81 def test_setup(self, mock__init_ix_scenario):
82 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
83 ixia_resource_helper.setup()
84 mock__init_ix_scenario.assert_called_once()
86 def test_stop_collect_with_client(self):
87 mock_client = mock.Mock()
88 ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
89 ixia_resource_helper.client = mock_client
90 ixia_resource_helper._ix_scenario = mock.Mock()
91 ixia_resource_helper.stop_collect()
92 self.assertEqual(1, ixia_resource_helper._terminated.value)
93 ixia_resource_helper._ix_scenario.stop_protocols.assert_called_once()
95 def test_run_traffic(self):
96 mock_tprofile = mock.Mock()
97 mock_tprofile.config.duration = 10
98 mock_tprofile.get_drop_percentage.return_value = True, 'fake_samples'
99 ixia_rhelper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
100 ixia_rhelper.rfc_helper = mock.Mock()
101 ixia_rhelper.vnfd_helper = mock.Mock()
102 ixia_rhelper._ix_scenario = mock.Mock()
103 ixia_rhelper.vnfd_helper.port_pairs.all_ports = []
104 with mock.patch.object(ixia_rhelper, 'generate_samples'), \
105 mock.patch.object(ixia_rhelper, '_build_ports'), \
106 mock.patch.object(ixia_rhelper, '_initialize_client'), \
107 mock.patch.object(utils, 'wait_until_true'):
108 ixia_rhelper.run_traffic(mock_tprofile)
110 self.assertEqual('fake_samples', ixia_rhelper._queue.get())
111 mock_tprofile.update_traffic_profile.assert_called_once()
113 def test_run_test(self):
114 expected_result = {'test': 'fake_samples', 'Iteration': 1}
115 mock_tprofile = mock.Mock()
116 mock_tprofile.config.duration = 10
117 mock_tprofile.get_drop_percentage.return_value = \
118 True, {'test': 'fake_samples', 'Iteration': 1}
119 ixia_rhelper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
120 tasks_queue = mock.Mock()
121 tasks_queue.get.return_value = 'RUN_TRAFFIC'
122 results_queue = mock.Mock()
123 ixia_rhelper.rfc_helper = mock.Mock()
124 ixia_rhelper.vnfd_helper = mock.Mock()
125 ixia_rhelper._ix_scenario = mock.Mock()
126 ixia_rhelper.vnfd_helper.port_pairs.all_ports = []
127 with mock.patch.object(ixia_rhelper, 'generate_samples'), \
128 mock.patch.object(ixia_rhelper, '_build_ports'), \
129 mock.patch.object(ixia_rhelper, '_initialize_client'), \
130 mock.patch.object(utils, 'wait_until_true'):
131 ixia_rhelper.run_test(mock_tprofile, tasks_queue, results_queue)
133 self.assertEqual(expected_result, ixia_rhelper._queue.get())
134 mock_tprofile.update_traffic_profile.assert_called_once()
135 tasks_queue.task_done.assert_called_once()
136 results_queue.put.assert_called_once_with('COMPLETE')
139 @mock.patch.object(tg_rfc2544_ixia, 'ixnet_api')
140 class TestIXIATrafficGen(unittest.TestCase):
141 VNFD = {'vnfd:vnfd-catalog':
143 [{'short-name': 'VpeVnf',
146 [{'network': '152.16.100.20',
147 'netmask': '255.255.255.0',
148 'gateway': '152.16.100.20',
150 {'network': '152.16.40.20',
151 'netmask': '255.255.255.0',
152 'gateway': '152.16.40.20',
154 'description': 'VPE approximation using DPDK',
155 'name': 'vpevnf-baremetal',
157 [{'network': '0064:ff9b:0:0:0:0:9810:6414',
159 'gateway': '0064:ff9b:0:0:0:0:9810:6414',
161 {'network': '0064:ff9b:0:0:0:0:9810:2814',
163 'gateway': '0064:ff9b:0:0:0:0:9810:2814',
165 'id': 'vpevnf-baremetal',
166 'external-interface':
167 [{'virtual-interface':
168 {'dst_mac': '00:00:00:00:00:04',
169 'vpci': '0000:05:00.0',
170 'local_ip': '152.16.100.19',
171 'type': 'PCI-PASSTHROUGH',
172 'netmask': '255.255.255.0',
174 'bandwidth': '10 Gbps',
176 'dst_ip': '152.16.100.20',
177 'local_iface_name': 'xe0',
178 'local_mac': '00:00:00:00:00:02'},
179 'vnfd-connection-point-ref': 'xe0',
181 {'virtual-interface':
182 {'dst_mac': '00:00:00:00:00:03',
183 'vpci': '0000:05:00.1',
184 'local_ip': '152.16.40.19',
185 'type': 'PCI-PASSTHROUGH',
187 'netmask': '255.255.255.0',
189 'bandwidth': '10 Gbps',
190 'dst_ip': '152.16.40.20',
191 'local_iface_name': 'xe1',
192 'local_mac': '00:00:00:00:00:01'},
193 'vnfd-connection-point-ref': 'xe1',
195 'description': 'Vpe approximation using DPDK',
197 {'vdu-id': 'vpevnf-baremetal',
203 {'kpi': ['packets_in', 'packets_fwd',
205 'connection-point': [{'type': 'VPORT', 'name': 'xe0'},
206 {'type': 'VPORT', 'name': 'xe1'}],
207 'id': 'VpeApproxVnf', 'name': 'VPEVnfSsh'}]}}
210 "schema": "isb:traffic_profile:0.1",
212 "description": "Fixed traffic profile to run UDP traffic",
214 "traffic_type": "FixedTraffic",
215 "frame_rate": 100, # pps
219 TC_YAML = {'scenarios': [{'tc_options':
220 {'rfc2544': {'allowed_drop_rate': '0.8 - 1'}},
221 'runner': {'duration': 400,
222 'interval': 35, 'type': 'Duration'},
224 {'flow': 'ipv4_1flow_Packets_vpe.yaml',
225 'imix': 'imix_voice.yaml'},
226 'vnf_options': {'vpe': {'cfg': 'vpe_config'}},
227 'traffic_profile': 'ipv4_throughput_vpe.yaml',
229 'nodes': {'tg__1': 'trafficgen_1.yardstick',
230 'vnf__1': 'vnf.yardstick'},
231 'topology': 'vpe_vnf_topology.yaml'}],
232 'context': {'nfvi_type': 'baremetal',
233 'type': contexts.CONTEXT_NODE,
235 'file': '/etc/yardstick/nodes/pod.yaml'},
236 'schema': 'yardstick:task:0.1'}
238 def test___init__(self, *args):
239 with mock.patch("yardstick.ssh.SSH") as ssh:
240 ssh_mock = mock.Mock(autospec=ssh.SSH)
242 mock.Mock(return_value=(0, "", ""))
243 ssh.from_node.return_value = ssh_mock
244 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
245 # NOTE(ralonsoh): check the object returned.
246 tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
248 def test_listen_traffic(self, *args):
249 with mock.patch("yardstick.ssh.SSH") as ssh:
250 ssh_mock = mock.Mock(autospec=ssh.SSH)
252 mock.Mock(return_value=(0, "", ""))
253 ssh.from_node.return_value = ssh_mock
254 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
255 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
256 self.assertIsNone(ixnet_traffic_gen.listen_traffic({}))
258 @mock.patch.object(ctx_base.Context, 'get_context_from_server', return_value='fake_context')
259 def test_instantiate(self, *args):
260 with mock.patch("yardstick.ssh.SSH") as ssh:
261 ssh_mock = mock.Mock(autospec=ssh.SSH)
263 mock.Mock(return_value=(0, "", ""))
265 mock.Mock(return_value=(0, "", ""))
266 ssh.from_node.return_value = ssh_mock
267 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
268 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
269 scenario_cfg = {'tc': "nsb_test_case",
277 'allowed_drop_rate': '0.8 - 1'},
279 'rules': 'acl_1rule.yaml',
283 'worker_config': '1C/1T',
284 'worker_threads': 1}}}})
285 scenario_cfg.update({
286 'nodes': {ixnet_traffic_gen.name: "mock"}
288 ixnet_traffic_gen.topology = ""
289 ixnet_traffic_gen.get_ixobj = mock.MagicMock()
290 ixnet_traffic_gen._ixia_traffic_gen = mock.MagicMock()
291 ixnet_traffic_gen._ixia_traffic_gen._connect = mock.Mock()
294 ixnet_traffic_gen.instantiate(scenario_cfg, {}))
296 @mock.patch.object(ctx_base.Context, 'get_physical_node_from_server', return_value='mock_node')
297 def test_collect_kpi(self, *args):
298 with mock.patch("yardstick.ssh.SSH") as ssh:
299 ssh_mock = mock.Mock(autospec=ssh.SSH)
301 mock.Mock(return_value=(0, "", ""))
302 ssh.from_node.return_value = ssh_mock
304 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
305 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
306 ixnet_traffic_gen.scenario_helper.scenario_cfg = {
307 'nodes': {ixnet_traffic_gen.name: "mock"}
309 ixnet_traffic_gen.data = {}
310 restult = ixnet_traffic_gen.collect_kpi()
312 expected = {'collect_stats': {},
313 'physical_node': 'mock_node'}
315 self.assertEqual(expected, restult)
317 def test_terminate(self, *args):
318 with mock.patch("yardstick.ssh.SSH") as ssh:
319 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
320 ssh_mock = mock.Mock(autospec=ssh.SSH)
322 mock.Mock(return_value=(0, "", ""))
323 ssh.from_node.return_value = ssh_mock
324 ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(
325 NAME, vnfd, resource_helper_type=mock.Mock())
326 ixnet_traffic_gen._terminated = mock.MagicMock()
327 ixnet_traffic_gen._terminated.value = 0
328 ixnet_traffic_gen._ixia_traffic_gen = mock.MagicMock()
329 ixnet_traffic_gen._ixia_traffic_gen.ix_stop_traffic = mock.Mock()
330 ixnet_traffic_gen._traffic_process = mock.MagicMock()
331 ixnet_traffic_gen._traffic_process.terminate = mock.Mock()
332 self.assertIsNone(ixnet_traffic_gen.terminate())
334 def _get_file_abspath(self, filename):
335 curr_path = os.path.dirname(os.path.abspath(__file__))
336 file_path = os.path.join(curr_path, filename)
339 def test__check_status(self, *args):
340 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
341 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
344 @mock.patch("yardstick.ssh.SSH")
345 def test_traffic_runner(self, mock_ssh, *args):
346 mock_traffic_profile = mock.Mock(autospec=tp_base.TrafficProfile)
347 mock_traffic_profile.get_traffic_definition.return_value = "64"
348 mock_traffic_profile.params = self.TRAFFIC_PROFILE
349 # traffic_profile.ports is standardized on port_num
350 mock_traffic_profile.ports = [0, 1]
352 mock_ssh_instance = mock.Mock(autospec=mock_ssh.SSH)
353 mock_ssh_instance.execute.return_value = 0, "", ""
354 mock_ssh_instance.run.return_value = 0, "", ""
356 mock_ssh.from_node.return_value = mock_ssh_instance
358 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
359 vnfd["mgmt-interface"].update({
361 "ixchassis": "1.1.1.1",
362 "py_bin_path": "/root",
368 for ifname in range(1):
369 name = "xe{}".format(ifname)
377 "Valid_Frames_Rx": 1000,
383 samples.update({"CurrentDropPercentage": 0.0})
388 "Rx_Rate_Kbps": [20, 20],
389 "Tx_Rate_Kbps": [20, 20],
390 "Rx_Rate_Mbps": [10, 10],
391 "Tx_Rate_Mbps": [10, 10],
392 "CurrentDropPercentage": [0, 0],
393 "RxThroughput": [10, 10],
394 "TxThroughput": [10, 10],
395 "Frames_Tx": [1000, 1000],
396 "in_packets": [1000, 1000],
397 "Valid_Frames_Rx": [1000, 1000],
398 "out_packets": [1000, 1000],
402 mock_traffic_profile.execute_traffic.return_value = [
403 'Completed', samples]
404 mock_traffic_profile.get_drop_percentage.return_value = [
405 'Completed', samples]
407 sut = tg_rfc2544_ixia.IxiaTrafficGen(name, vnfd)
408 sut.vnf_port_pairs = [[[0], [1]]]
409 sut.tc_file_name = self._get_file_abspath(TEST_FILE_YAML)
412 sut.ssh_helper = mock.Mock()
413 sut._traffic_process = mock.MagicMock()
414 sut.generate_port_pairs = mock.Mock()
416 sut._ixia_traffic_gen = mock.MagicMock()
417 sut._ixia_traffic_gen.ix_get_statistics.return_value = last_res
419 sut.resource_helper.client = mock.MagicMock()
420 sut.resource_helper.client_started = mock.MagicMock()
421 sut.resource_helper.client_started.value = 1
422 sut.resource_helper.rfc_helper.iteration.value = 11
423 sut.resource_helper._ix_scenario = mock.Mock()
425 sut.scenario_helper.scenario_cfg = {
430 'allowed_drop_rate': '0.8 - 1',
434 'rules': 'acl_1rule.yaml',
438 'worker_config': '1C/1T',
443 'task_path': '/path/to/task'
446 @mock.patch.object(six.moves.builtins, 'open', create=True)
447 @mock.patch('yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.open',
448 mock.mock_open(), create=True)
449 @mock.patch('yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.LOG.exception')
450 def _traffic_runner(*args):
451 result = sut._traffic_runner(mock_traffic_profile)
452 self.assertIsNone(result)
456 def test_run_traffic_once(self, *args):
457 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
458 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
459 sut._init_traffic_process = mock.Mock()
460 sut._tasks_queue.put = mock.Mock()
461 sut.resource_helper.client_started.value = 0
462 sut.run_traffic_once(self.TRAFFIC_PROFILE)
463 sut._tasks_queue.put.assert_called_once_with("RUN_TRAFFIC")
464 sut._init_traffic_process.assert_called_once_with(self.TRAFFIC_PROFILE)
466 def test__test_runner(self, *args):
467 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
468 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
471 sut.resource_helper = mock.Mock()
472 sut._test_runner(self.TRAFFIC_PROFILE, tasks, results)
473 sut.resource_helper.run_test.assert_called_once_with(self.TRAFFIC_PROFILE,
476 @mock.patch.object(time, 'sleep', return_value=0)
477 def test__init_traffic_process(self, *args):
478 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
479 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
480 sut._test_runner = mock.Mock(return_value=0)
481 sut.resource_helper = mock.Mock()
482 sut.resource_helper.client_started.value = 0
483 sut._init_traffic_process(self.TRAFFIC_PROFILE)
485 def test_wait_on_traffic(self, *args):
486 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
487 sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
488 sut._tasks_queue.join = mock.Mock(return_value=0)
489 sut._result_queue.get = mock.Mock(return_value='COMPLETE')
490 result = sut.wait_on_traffic()
491 sut._tasks_queue.join.assert_called_once()
492 sut._result_queue.get.assert_called_once()
493 self.assertEqual(result, 'COMPLETE')
496 class TestIxiaBasicScenario(unittest.TestCase):
498 STATS = {'stat_name': ['Card01/Port01',
500 'port_name': ['Ethernet - 001', 'Ethernet - 002'],
501 'Frames_Tx': ['150', '150'],
502 'Valid_Frames_Rx': ['150', '150'],
503 'Frames_Tx_Rate': ['0.0', '0.0'],
504 'Valid_Frames_Rx_Rate': ['0.0', '0.0'],
505 'Bytes_Rx': ['9600', '9600'],
506 'Bytes_Tx': ['9600', '9600'],
507 'Tx_Rate_Kbps': ['0.0', '0.0'],
508 'Rx_Rate_Mbps': ['0.0', '0.0'],
509 'Tx_Rate_Mbps': ['0.0', '0.0'],
510 'Rx_Rate_Kbps': ['0.0', '0.0'],
511 'Store-Forward_Max_latency_ns': ['100', '200'],
512 'Store-Forward_Min_latency_ns': ['100', '200'],
513 'Store-Forward_Avg_latency_ns': ['100', '200']}
516 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
517 self.mock_IxNextgen = self._mock_IxNextgen.start()
518 self.context_cfg = mock.Mock()
519 self.ixia_cfg = mock.Mock()
520 self.scenario = tg_rfc2544_ixia.IxiaBasicScenario(self.mock_IxNextgen,
523 self.addCleanup(self._stop_mocks)
525 def _stop_mocks(self):
526 self._mock_IxNextgen.stop()
528 def test___init___(self):
529 self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaBasicScenario)
530 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
532 def test_create_traffic_model(self):
533 self.mock_IxNextgen.get_vports.return_value = [1, 2, 3, 4]
534 yaml_data = {'traffic_profile': {}
536 traffic_profile = ixia_rfc2544.IXIARFC2544Profile(yaml_data)
537 self.scenario.create_traffic_model(traffic_profile)
538 self.scenario.client.get_vports.assert_called_once()
539 self.scenario.client.create_traffic_model.assert_called_once_with(
540 [1, 3], [2, 4], traffic_profile)
542 def test_apply_config(self):
543 self.assertIsNone(self.scenario.apply_config())
545 def test_run_protocols(self):
546 self.assertIsNone(self.scenario.run_protocols())
548 def test_stop_protocols(self):
549 self.assertIsNone(self.scenario.stop_protocols())
551 def test__get_stats(self):
552 self.scenario._get_stats()
553 self.scenario.client.get_statistics.assert_called_once()
555 @mock.patch.object(tg_rfc2544_ixia.IxiaBasicScenario, '_get_stats')
556 def test_generate_samples(self, mock_get_stats):
558 expected_samples = {'xe0': {
565 'RxThroughputBps': 320.0,
566 'TxThroughputBps': 320.0,
577 'RxThroughputBps': 320.0,
578 'TxThroughputBps': 320.0,
583 res_helper = mock.Mock()
584 res_helper.vnfd_helper.find_interface_by_port.side_effect = \
585 [{'name': 'xe0'}, {'name': 'xe1'}]
588 mock_get_stats.return_value = self.STATS
589 samples = self.scenario.generate_samples(res_helper, ports, duration)
590 mock_get_stats.assert_called_once()
591 self.assertEqual(samples, expected_samples)
594 class TestIxiaL3Scenario(TestIxiaBasicScenario):
597 'src_ip': ['192.168.0.1-192.168.0.50'],
598 'dst_ip': ['192.168.1.1-192.168.1.150']
608 'vld_id': 'uplink_0',
609 'local_ip': '10.1.1.1',
610 'local_mac': 'aa:bb:cc:dd:ee:ff',
614 'vld_id': 'downlink_0',
615 'local_ip': '20.2.2.2',
616 'local_mac': 'bb:bb:cc:dd:ee:ee',
621 'network': "152.16.100.20",
622 'netmask': '255.255.0.0',
623 'gateway': '152.16.100.21',
631 super(TestIxiaL3Scenario, self).setUp()
632 self.ixia_cfg = self.IXIA_CFG
633 self.context_cfg = self.CONTEXT_CFG
634 self.scenario = tg_rfc2544_ixia.IxiaL3Scenario(self.mock_IxNextgen,
638 def test___init___(self):
639 self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaL3Scenario)
640 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
642 def test_create_traffic_model(self):
643 self.mock_IxNextgen.get_vports.return_value = ['1', '2']
644 traffic_profile = 'fake_profile'
645 self.scenario.create_traffic_model(traffic_profile)
646 self.scenario.client.get_vports.assert_called_once()
647 self.scenario.client.create_ipv4_traffic_model.\
648 assert_called_once_with(['1/protocols/static'],
649 ['2/protocols/static'],
652 def test_apply_config(self):
653 self.scenario._add_interfaces = mock.Mock()
654 self.scenario._add_static_ips = mock.Mock()
655 self.assertIsNone(self.scenario.apply_config())
657 def test__add_static(self):
658 self.mock_IxNextgen.get_vports.return_value = ['1', '2']
659 self.mock_IxNextgen.get_static_interface.side_effect = ['intf1',
662 self.scenario._add_static_ips()
664 self.mock_IxNextgen.get_static_interface.assert_any_call('1')
665 self.mock_IxNextgen.get_static_interface.assert_any_call('2')
667 self.scenario.client.add_static_ipv4.assert_any_call(
668 'intf1', '1', '192.168.0.1', 49, '32')
669 self.scenario.client.add_static_ipv4.assert_any_call(
670 'intf2', '2', '192.168.1.1', 149, '32')
672 def test__add_interfaces(self):
673 self.mock_IxNextgen.get_vports.return_value = ['1', '2']
675 self.scenario._add_interfaces()
677 self.mock_IxNextgen.add_interface.assert_any_call('1',
681 self.mock_IxNextgen.add_interface.assert_any_call('2',
687 class TestIxiaPppoeClientScenario(unittest.TestCase):
691 'sessions_per_port': 4,
692 'sessions_per_svlan': 1,
695 'ip': ['10.3.3.1', '10.4.4.1']
698 'sessions_per_port': 1,
699 'sessions_per_vlan': 1,
701 'gateway_ip': ['10.1.1.1', '10.2.2.1'],
702 'ip': ['10.1.1.1', '10.2.2.1'],
703 'prefix': ['24', '24']
706 'tos': {'precedence': [0, 4]}
712 'interfaces': {'xe0': {
713 'local_ip': '10.1.1.1',
714 'netmask': '255.255.255.0'
718 self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen')
719 self.mock_IxNextgen = self._mock_IxNextgen.start()
720 self.scenario = tg_rfc2544_ixia.IxiaPppoeClientScenario(
721 self.mock_IxNextgen, self.CONTEXT_CFG, self.IXIA_CFG)
722 tg_rfc2544_ixia.WAIT_PROTOCOLS_STARTED = 2
723 self.addCleanup(self._stop_mocks)
725 def _stop_mocks(self):
726 self._mock_IxNextgen.stop()
728 def test___init___(self):
729 self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaPppoeClientScenario)
730 self.assertEqual(self.scenario.client, self.mock_IxNextgen)
732 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
734 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
735 '_apply_access_network_config')
736 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
737 '_apply_core_network_config')
738 def test_apply_config(self, mock_apply_core_net_cfg,
739 mock_apply_access_net_cfg,
740 mock_fill_ixia_config):
741 self.mock_IxNextgen.get_vports.return_value = [1, 2, 3, 4]
742 self.scenario.apply_config()
743 self.scenario.client.get_vports.assert_called_once()
744 self.assertEqual(self.scenario._uplink_vports, [1, 3])
745 self.assertEqual(self.scenario._downlink_vports, [2, 4])
746 mock_fill_ixia_config.assert_called_once()
747 mock_apply_core_net_cfg.assert_called_once()
748 mock_apply_access_net_cfg.assert_called_once()
750 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
751 '_get_endpoints_src_dst_id_pairs')
752 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
753 '_get_endpoints_src_dst_obj_pairs')
754 def test_create_traffic_model(self, mock_obj_pairs, mock_id_pairs):
755 uplink_endpoints = ['group1', 'group2']
756 downlink_endpoints = ['group3', 'group3']
757 mock_id_pairs.return_value = ['xe0', 'xe1', 'xe0', 'xe1']
758 mock_obj_pairs.return_value = ['group1', 'group3', 'group2', 'group3']
759 mock_tp = mock.Mock()
760 mock_tp.full_profile = {'uplink_0': 'data',
761 'downlink_0': 'data',
765 self.scenario.create_traffic_model(mock_tp)
766 mock_id_pairs.assert_called_once_with(mock_tp.full_profile)
767 mock_obj_pairs.assert_called_once_with(['xe0', 'xe1', 'xe0', 'xe1'])
768 self.scenario.client.create_ipv4_traffic_model.assert_called_once_with(
769 uplink_endpoints, downlink_endpoints, mock_tp)
771 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
772 '_get_endpoints_src_dst_id_pairs')
773 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
774 '_get_endpoints_src_dst_obj_pairs')
775 def test_create_traffic_model_topology_based_flows(self, mock_obj_pairs,
777 uplink_topologies = ['topology1', 'topology3']
778 downlink_topologies = ['topology2', 'topology4']
779 mock_id_pairs.return_value = []
780 mock_obj_pairs.return_value = []
781 mock_tp = mock.Mock()
782 mock_tp.full_profile = {'uplink_0': 'data',
783 'downlink_0': 'data',
787 self.scenario._access_topologies = ['topology1', 'topology3']
788 self.scenario._core_topologies = ['topology2', 'topology4']
789 self.scenario.create_traffic_model(mock_tp)
790 mock_id_pairs.assert_called_once_with(mock_tp.full_profile)
791 mock_obj_pairs.assert_called_once_with([])
792 self.scenario.client.create_ipv4_traffic_model.assert_called_once_with(
793 uplink_topologies, downlink_topologies, mock_tp)
795 def test__get_endpoints_src_dst_id_pairs(self):
796 full_tp = OrderedDict([
797 ('uplink_0', {'ipv4': {'port': 'xe0'}}),
798 ('downlink_0', {'ipv4': {'port': 'xe1'}}),
799 ('uplink_1', {'ipv4': {'port': 'xe0'}}),
800 ('downlink_1', {'ipv4': {'port': 'xe3'}})])
801 endpoints_src_dst_pairs = ['xe0', 'xe1', 'xe0', 'xe3']
802 res = self.scenario._get_endpoints_src_dst_id_pairs(full_tp)
803 self.assertEqual(res, endpoints_src_dst_pairs)
805 def test__get_endpoints_src_dst_id_pairs_wrong_flows_number(self):
806 full_tp = OrderedDict([
807 ('uplink_0', {'ipv4': {'port': 'xe0'}}),
808 ('downlink_0', {'ipv4': {'port': 'xe1'}}),
809 ('uplink_1', {'ipv4': {'port': 'xe0'}})])
810 with self.assertRaises(RuntimeError):
811 self.scenario._get_endpoints_src_dst_id_pairs(full_tp)
813 def test__get_endpoints_src_dst_id_pairs_no_port_key(self):
814 full_tp = OrderedDict([
815 ('uplink_0', {'ipv4': {'id': 1}}),
816 ('downlink_0', {'ipv4': {'id': 2}})])
818 self.scenario._get_endpoints_src_dst_id_pairs(full_tp), [])
820 def test__get_endpoints_src_dst_obj_pairs_tp_with_port_key(self):
821 endpoints_id_pairs = ['xe0', 'xe1',
827 'sessions_per_port': 4,
828 'sessions_per_svlan': 1
831 'src_ip': [{'tg__0': 'xe0'}, {'tg__0': 'xe2'}],
832 'dst_ip': [{'tg__0': 'xe1'}, {'tg__0': 'xe3'}]
836 expected_result = ['tp1_dg1', 'tp3_dg1', 'tp1_dg2', 'tp3_dg1',
837 'tp1_dg3', 'tp4_dg1', 'tp1_dg4', 'tp4_dg1']
839 self.scenario._ixia_cfg = ixia_cfg
840 self.scenario._access_topologies = ['topology1', 'topology2']
841 self.scenario._core_topologies = ['topology3', 'topology4']
842 self.mock_IxNextgen.get_topology_device_groups.side_effect = \
843 [['tp1_dg1', 'tp1_dg2', 'tp1_dg3', 'tp1_dg4'],
844 ['tp2_dg1', 'tp2_dg2', 'tp2_dg3', 'tp2_dg4'],
847 res = self.scenario._get_endpoints_src_dst_obj_pairs(
849 self.assertEqual(res, expected_result)
851 def test__get_endpoints_src_dst_obj_pairs_default_flows_mapping(self):
852 endpoints_id_pairs = []
855 'sessions_per_port': 4,
856 'sessions_per_svlan': 1
859 'src_ip': [{'tg__0': 'xe0'}, {'tg__0': 'xe2'}],
860 'dst_ip': [{'tg__0': 'xe1'}, {'tg__0': 'xe3'}]
864 self.scenario._ixia_cfg = ixia_cfg
865 res = self.scenario._get_endpoints_src_dst_obj_pairs(
867 self.assertEqual(res, [])
869 def test_run_protocols(self):
870 self.scenario.client.is_protocols_running.return_value = True
871 self.scenario.run_protocols()
872 self.scenario.client.start_protocols.assert_called_once()
874 def test_run_protocols_timeout_exception(self):
875 self.scenario.client.is_protocols_running.return_value = False
876 with self.assertRaises(exceptions.WaitTimeout):
877 self.scenario.run_protocols()
878 self.scenario.client.start_protocols.assert_called_once()
880 def test_stop_protocols(self):
881 self.scenario.stop_protocols()
882 self.scenario.client.stop_protocols.assert_called_once()
884 def test__get_intf_addr_str_type_input(self):
885 intf = '192.168.10.2/24'
886 ip, mask = self.scenario._get_intf_addr(intf)
887 self.assertEqual(ip, '192.168.10.2')
888 self.assertEqual(mask, 24)
890 def test__get_intf_addr_dict_type_input(self):
891 intf = {'tg__0': 'xe0'}
892 ip, mask = self.scenario._get_intf_addr(intf)
893 self.assertEqual(ip, '10.1.1.1')
894 self.assertEqual(mask, 24)
896 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario, '_get_intf_addr')
897 def test__fill_ixia_config(self, mock_get_intf_addr):
901 'sessions_per_port': 4,
902 'sessions_per_svlan': 1,
905 'ip': ['10.3.3.1/24', '10.4.4.1/24']
908 'sessions_per_port': 1,
909 'sessions_per_vlan': 1,
911 'gateway_ip': ['10.1.1.1/24', '10.2.2.1/24'],
912 'ip': ['10.1.1.1/24', '10.2.2.1/24']
916 mock_get_intf_addr.side_effect = [
924 self.scenario._ixia_cfg = ixia_cfg
925 self.scenario._fill_ixia_config()
926 self.assertEqual(mock_get_intf_addr.call_count, 6)
927 self.assertEqual(self.scenario._ixia_cfg['pppoe_client']['ip'],
928 ['10.3.3.1', '10.4.4.1'])
929 self.assertEqual(self.scenario._ixia_cfg['ipv4_client']['ip'],
930 ['10.1.1.1', '10.2.2.1'])
931 self.assertEqual(self.scenario._ixia_cfg['ipv4_client']['prefix'],
934 @mock.patch('yardstick.network_services.libs.ixia_libs.ixnet.ixnet_api.Vlan')
935 def test__apply_access_network_config_pap_auth(self, mock_vlan):
938 'sessions_per_port': 4,
939 'sessions_per_svlan': 1,
942 'pap_user': 'test_pap',
943 'pap_password': 'pap'
945 pap_user = _ixia_cfg['pppoe_client']['pap_user']
946 pap_passwd = _ixia_cfg['pppoe_client']['pap_password']
947 self.scenario._ixia_cfg = _ixia_cfg
948 self.scenario._uplink_vports = [0, 2]
949 self.scenario.client.add_topology.side_effect = ['Topology 1', 'Topology 2']
950 self.scenario.client.add_device_group.side_effect = ['Dg1', 'Dg2', 'Dg3',
953 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2', 'Eth3',
954 'Eth4', 'Eth5', 'Eth6',
956 self.scenario._apply_access_network_config()
957 self.assertEqual(self.scenario.client.add_topology.call_count, 2)
958 self.assertEqual(self.scenario.client.add_device_group.call_count, 8)
959 self.assertEqual(self.scenario.client.add_ethernet.call_count, 8)
960 self.assertEqual(mock_vlan.call_count, 16)
961 self.assertEqual(self.scenario.client.add_vlans.call_count, 8)
962 self.assertEqual(self.scenario.client.add_pppox_client.call_count, 8)
963 self.scenario.client.add_topology.assert_has_calls([
964 mock.call('Topology access 0', 0),
965 mock.call('Topology access 1', 2)
967 self.scenario.client.add_device_group.assert_has_calls([
968 mock.call('Topology 1', 'SVLAN 10', 1),
969 mock.call('Topology 1', 'SVLAN 11', 1),
970 mock.call('Topology 1', 'SVLAN 12', 1),
971 mock.call('Topology 1', 'SVLAN 13', 1),
972 mock.call('Topology 2', 'SVLAN 14', 1),
973 mock.call('Topology 2', 'SVLAN 15', 1),
974 mock.call('Topology 2', 'SVLAN 16', 1),
975 mock.call('Topology 2', 'SVLAN 17', 1)
977 self.scenario.client.add_ethernet.assert_has_calls([
978 mock.call('Dg1', 'Ethernet'),
979 mock.call('Dg2', 'Ethernet'),
980 mock.call('Dg3', 'Ethernet'),
981 mock.call('Dg4', 'Ethernet'),
982 mock.call('Dg5', 'Ethernet'),
983 mock.call('Dg6', 'Ethernet'),
984 mock.call('Dg7', 'Ethernet'),
985 mock.call('Dg8', 'Ethernet')
987 mock_vlan.assert_has_calls([
988 mock.call(vlan_id=10),
989 mock.call(vlan_id=20, vlan_id_step=1),
990 mock.call(vlan_id=11),
991 mock.call(vlan_id=20, vlan_id_step=1),
992 mock.call(vlan_id=12),
993 mock.call(vlan_id=20, vlan_id_step=1),
994 mock.call(vlan_id=13),
995 mock.call(vlan_id=20, vlan_id_step=1),
996 mock.call(vlan_id=14),
997 mock.call(vlan_id=20, vlan_id_step=1),
998 mock.call(vlan_id=15),
999 mock.call(vlan_id=20, vlan_id_step=1),
1000 mock.call(vlan_id=16),
1001 mock.call(vlan_id=20, vlan_id_step=1),
1002 mock.call(vlan_id=17),
1003 mock.call(vlan_id=20, vlan_id_step=1)
1005 self.scenario.client.add_pppox_client.assert_has_calls([
1006 mock.call('Eth1', 'pap', pap_user, pap_passwd),
1007 mock.call('Eth2', 'pap', pap_user, pap_passwd),
1008 mock.call('Eth3', 'pap', pap_user, pap_passwd),
1009 mock.call('Eth4', 'pap', pap_user, pap_passwd),
1010 mock.call('Eth5', 'pap', pap_user, pap_passwd),
1011 mock.call('Eth6', 'pap', pap_user, pap_passwd),
1012 mock.call('Eth7', 'pap', pap_user, pap_passwd),
1013 mock.call('Eth8', 'pap', pap_user, pap_passwd)
1016 def test__apply_access_network_config_chap_auth(self):
1019 'sessions_per_port': 4,
1020 'sessions_per_svlan': 1,
1023 'chap_user': 'test_chap',
1024 'chap_password': 'chap'
1026 chap_user = _ixia_cfg['pppoe_client']['chap_user']
1027 chap_passwd = _ixia_cfg['pppoe_client']['chap_password']
1028 self.scenario._ixia_cfg = _ixia_cfg
1029 self.scenario._uplink_vports = [0, 2]
1030 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2', 'Eth3',
1031 'Eth4', 'Eth5', 'Eth6',
1033 self.scenario._apply_access_network_config()
1034 self.assertEqual(self.scenario.client.add_pppox_client.call_count, 8)
1035 self.scenario.client.add_pppox_client.assert_has_calls([
1036 mock.call('Eth1', 'chap', chap_user, chap_passwd),
1037 mock.call('Eth2', 'chap', chap_user, chap_passwd),
1038 mock.call('Eth3', 'chap', chap_user, chap_passwd),
1039 mock.call('Eth4', 'chap', chap_user, chap_passwd),
1040 mock.call('Eth5', 'chap', chap_user, chap_passwd),
1041 mock.call('Eth6', 'chap', chap_user, chap_passwd),
1042 mock.call('Eth7', 'chap', chap_user, chap_passwd),
1043 mock.call('Eth8', 'chap', chap_user, chap_passwd)
1046 @mock.patch('yardstick.network_services.libs.ixia_libs.ixnet.ixnet_api.Vlan')
1047 def test__apply_core_network_config_no_bgp_proto(self, mock_vlan):
1048 self.scenario._downlink_vports = [1, 3]
1049 self.scenario.client.add_topology.side_effect = ['Topology 1', 'Topology 2']
1050 self.scenario.client.add_device_group.side_effect = ['Dg1', 'Dg2']
1051 self.scenario.client.add_ethernet.side_effect = ['Eth1', 'Eth2']
1052 self.scenario._apply_core_network_config()
1053 self.assertEqual(self.scenario.client.add_topology.call_count, 2)
1054 self.assertEqual(self.scenario.client.add_device_group.call_count, 2)
1055 self.assertEqual(self.scenario.client.add_ethernet.call_count, 2)
1056 self.assertEqual(mock_vlan.call_count, 2)
1057 self.assertEqual(self.scenario.client.add_vlans.call_count, 2)
1058 self.assertEqual(self.scenario.client.add_ipv4.call_count, 2)
1059 self.scenario.client.add_topology.assert_has_calls([
1060 mock.call('Topology core 0', 1),
1061 mock.call('Topology core 1', 3)
1063 self.scenario.client.add_device_group.assert_has_calls([
1064 mock.call('Topology 1', 'Core port 0', 1),
1065 mock.call('Topology 2', 'Core port 1', 1)
1067 self.scenario.client.add_ethernet.assert_has_calls([
1068 mock.call('Dg1', 'Ethernet'),
1069 mock.call('Dg2', 'Ethernet')
1071 mock_vlan.assert_has_calls([
1072 mock.call(vlan_id=101),
1073 mock.call(vlan_id=102)
1075 self.scenario.client.add_ipv4.assert_has_calls([
1076 mock.call('Eth1', name='ipv4', addr=ipaddress.IPv4Address('10.1.1.2'),
1077 addr_step='0.0.0.1', prefix='24', gateway='10.1.1.1'),
1078 mock.call('Eth2', name='ipv4', addr=ipaddress.IPv4Address('10.2.2.2'),
1079 addr_step='0.0.0.1', prefix='24', gateway='10.2.2.1')
1081 self.scenario.client.add_bgp.assert_not_called()
1083 def test__apply_core_network_config_with_bgp_proto(self):
1086 'bgp_type': 'external',
1087 'dut_ip': '10.0.0.1',
1091 self.scenario._ixia_cfg['ipv4_client'].update(bgp_params)
1092 self.scenario._downlink_vports = [1, 3]
1093 self.scenario.client.add_ipv4.side_effect = ['ipv4_1', 'ipv4_2']
1094 self.scenario._apply_core_network_config()
1095 self.assertEqual(self.scenario.client.add_bgp.call_count, 2)
1096 self.scenario.client.add_bgp.assert_has_calls([
1097 mock.call('ipv4_1', dut_ip=bgp_params["bgp"]["dut_ip"],
1098 local_as=bgp_params["bgp"]["as_number"],
1099 bgp_type=bgp_params["bgp"]["bgp_type"]),
1100 mock.call('ipv4_2', dut_ip=bgp_params["bgp"]["dut_ip"],
1101 local_as=bgp_params["bgp"]["as_number"],
1102 bgp_type=bgp_params["bgp"]["bgp_type"])
1105 def test_update_tracking_options_raw_priority(self):
1106 raw_priority = {'raw': 4}
1107 self.scenario._ixia_cfg['priority'] = raw_priority
1108 self.scenario.update_tracking_options()
1109 self.scenario.client.set_flow_tracking.assert_called_once_with(
1110 ['flowGroup0', 'vlanVlanId0', 'ipv4Raw0'])
1112 def test_update_tracking_options_tos_priority(self):
1113 tos_priority = {'tos': {'precedence': [4, 7]}}
1114 self.scenario._ixia_cfg['priority'] = tos_priority
1115 self.scenario.update_tracking_options()
1116 self.scenario.client.set_flow_tracking.assert_called_once_with(
1117 ['flowGroup0', 'vlanVlanId0', 'ipv4Precedence0'])
1119 def test_update_tracking_options_dscp_priority(self):
1120 dscp_priority = {'dscp': {'defaultPHB': [4, 7]}}
1121 self.scenario._ixia_cfg['priority'] = dscp_priority
1122 self.scenario.update_tracking_options()
1123 self.scenario.client.set_flow_tracking.assert_called_once_with(
1124 ['flowGroup0', 'vlanVlanId0', 'ipv4DefaultPhb0'])
1126 def test_update_tracking_options_invalid_priority_data(self):
1127 invalid_priority = {'tos': {'inet-precedence': [4, 7]}}
1128 self.scenario._ixia_cfg['priority'] = invalid_priority
1129 self.scenario.update_tracking_options()
1130 self.scenario.client.set_flow_tracking.assert_called_once_with(
1131 ['flowGroup0', 'vlanVlanId0', 'ipv4Precedence0'])
1133 def test_get_tc_rfc2544_options(self):
1134 rfc2544_tc_opts = {'allowed_drop_rate': '0.0001 - 0.0001'}
1135 self.scenario._ixia_cfg['rfc2544'] = rfc2544_tc_opts
1136 res = self.scenario.get_tc_rfc2544_options()
1137 self.assertEqual(res, rfc2544_tc_opts)
1139 def test__get_stats(self):
1140 self.scenario._get_stats()
1141 self.scenario.client.get_pppoe_scenario_statistics.assert_called_once()
1143 def test_get_flow_id_data(self):
1144 stats = [{'id': 1, 'in_packets': 10, 'out_packets': 20}]
1147 res = self.scenario.get_flow_id_data(stats, flow_id, key)
1148 self.assertEqual(res, 10)
1150 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario, '_get_stats')
1151 @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario,
1152 'get_priority_flows_stats')
1153 def test_generate_samples(self, mock_prio_flow_statistics,
1157 {'Flow_Group': 'RFC2544-1 - Flow Group 0001',
1158 'Frames_Delta': '0',
1160 'Rx_Frames': '3000',
1161 'Tx_Frames': '3000',
1163 'Tx_Port': 'Ethernet - 001',
1164 'Store-Forward_Avg_latency_ns': '2',
1165 'Store-Forward_Min_latency_ns': '2',
1166 'Store-Forward_Max_latency_ns': '2'},
1167 {'Flow_Group': 'RFC2544-2 - Flow Group 0001',
1168 'Frames_Delta': '0',
1170 'Rx_Frames': '3000',
1171 'Tx_Frames': '3000',
1173 'Tx_Port': 'Ethernet - 002',
1174 'Store-Forward_Avg_latency_ns': '2',
1175 'Store-Forward_Min_latency_ns': '2',
1176 'Store-Forward_Max_latency_ns': '2'
1178 'port_statistics': [
1179 {'Frames_Tx': '3000',
1180 'Valid_Frames_Rx': '3000',
1181 'Bytes_Rx': '192000',
1182 'Bytes_Tx': '192000',
1183 'Rx_Rate_Kbps': '0.0',
1184 'Tx_Rate_Kbps': '0.0',
1185 'Rx_Rate_Mbps': '0.0',
1186 'Tx_Rate_Mbps': '0.0',
1187 'port_name': 'Ethernet - 001'},
1188 {'Frames_Tx': '3000',
1189 'Valid_Frames_Rx': '3000',
1190 'Bytes_Rx': '192000',
1191 'Bytes_Tx': '192000',
1192 'Rx_Rate_Kbps': '0.0',
1193 'Tx_Rate_Kbps': '0.0',
1194 'Rx_Rate_Mbps': '0.0',
1195 'Tx_Rate_Mbps': '0.0',
1196 'port_name': 'Ethernet - 002'}],
1197 'pppox_client_per_port': [
1198 {'Sessions_Down': '0',
1199 'Sessions_Not_Started': '0',
1200 'Sessions_Total': '1',
1202 'subs_port': 'Ethernet - 001'}]}
1204 prio_flows_stats = {
1208 'RxThroughput': 200.0,
1209 'TxThroughput': 200.0,
1216 expected_result = {'priority_stats': {
1217 '0': {'RxThroughput': 200.0,
1218 'TxThroughput': 200.0,
1223 'OutPackets': 6000}},
1224 'xe0': {'RxThroughput': 100.0,
1228 'TxThroughput': 100.0,
1233 'RxThroughputBps': 6400.0,
1234 'TxThroughputBps': 6400.0,
1236 'SessionsNotStarted': 0,
1239 'xe1': {'RxThroughput': 100.0,
1243 'TxThroughput': 100.0,
1248 'RxThroughputBps': 6400.0,
1249 'TxThroughputBps': 6400.0}}
1251 mock_get_stats.return_value = ixia_stats
1252 mock_prio_flow_statistics.return_value = prio_flows_stats
1254 port_names = [{'name': 'xe0'}, {'name': 'xe1'}]
1256 res_helper = mock.Mock()
1257 res_helper.vnfd_helper.find_interface_by_port.side_effect = \
1259 samples = self.scenario.generate_samples(res_helper, ports, duration)
1260 self.assertIsNotNone(samples)
1261 self.assertIsNotNone(samples.get('xe0'))
1262 self.assertIsNotNone(samples.get('xe1'))
1263 self.assertEqual(samples, expected_result)
1264 mock_get_stats.assert_called_once()
1265 mock_prio_flow_statistics.assert_called_once()