- "frame_size": 64}}
-
- TC_YAML = {'scenarios': [{'tc_options':
- {'rfc2544': {'allowed_drop_rate': '0.8 - 1'}},
- 'runner': {'duration': 400,
- 'interval': 35, 'type': 'Duration'},
- 'traffic_options':
- {'flow': 'ipv4_1flow_Packets_vpe.yaml',
- 'imix': 'imix_voice.yaml'},
- 'vnf_options': {'vpe': {'cfg': 'vpe_config'}},
- 'traffic_profile': 'ipv4_throughput_vpe.yaml',
- 'type': 'NSPerf',
- 'nodes': {'tg__1': 'trafficgen_1.yardstick',
- 'vnf__1': 'vnf.yardstick'},
- 'topology': 'vpe_vnf_topology.yaml'}],
- 'context': {'nfvi_type': 'baremetal', 'type': 'Node',
- 'name': 'yardstick',
- 'file': '/etc/yardstick/nodes/pod.yaml'},
- 'schema': 'yardstick:task:0.1'}
-
- def test___init__(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = TrexTrafficGenRFC(vnfd)
- self.assertIsNotNone(trex_traffic_gen._terminated)
-
- def test_collect_kpi(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = TrexTrafficGenRFC(vnfd)
- restult = trex_traffic_gen.collect_kpi()
- self.assertEqual({}, restult)
-
- def test_listen_traffic(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = TrexTrafficGenRFC(vnfd)
- self.assertEqual(None, trex_traffic_gen.listen_traffic({}))
-
- def test_instantiate(self):
- mock_traffic_profile = mock.Mock(autospec=TrafficProfile)
- mock_traffic_profile.get_traffic_definition.return_value = "64"
- mock_traffic_profile.params = self.TRAFFIC_PROFILE
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh_mock.run = \
- mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = TrexTrafficGenRFC(vnfd)
- trex_traffic_gen._start_server = mock.Mock(return_value=0)
- scenario_cfg = {"tc": "tc_baremetal_rfc2544_ipv4_1flow_64B"}
- tg_rfc2544_trex.WAIT_TIME = 3
- self.assertEqual(0, trex_traffic_gen.instantiate(scenario_cfg, {}))
-
- def test_instantiate_error(self):
- mock_traffic_profile = mock.Mock(autospec=TrafficProfile)
- mock_traffic_profile.get_traffic_definition.return_value = "64"
- mock_traffic_profile.params = self.TRAFFIC_PROFILE
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(1, "", ""))
- ssh_mock.run = \
- mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = TrexTrafficGenRFC(vnfd)
- scenario_cfg = {"tc": "tc_baremetal_rfc2544_ipv4_1flow_64B"}
- self.assertRaises(RuntimeError,
- trex_traffic_gen.instantiate, scenario_cfg, {})
-
- def test__get_rfc_tolerance(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh_mock.run = \
- mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = TrexTrafficGenRFC(vnfd)
- self.assertEqual([0.8, 1.0],
- trex_traffic_gen._get_rfc_tolerance(self.TC_YAML))
- self.TC_YAML["scenarios"][0]["tc_options"]['rfc2544'][
- 'allowed_drop_rate'] = '0.8'
- self.assertEqual([0.8, 0.8],
- trex_traffic_gen._get_rfc_tolerance(self.TC_YAML))
-
- def test__start_server(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh_mock.run = \
- mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = TrexTrafficGenRFC(vnfd)
- self.assertEqual(None, trex_traffic_gen._start_server())
-
- def _get_file_abspath(self, filename):
- curr_path = os.path.dirname(os.path.abspath(__file__))
- file_path = os.path.join(curr_path, filename)
- return file_path
-
- def test__traffic_runner(self):
+ "frame_size": 64,
+ },
+ }
+
+ TC_YAML = {
+ 'scenarios': [
+ {
+ 'tc_options': {
+ 'rfc2544': {
+ 'allowed_drop_rate': '0.8 - 1',
+ },
+ },
+ 'runner': {
+ 'duration': 400,
+ 'interval': 35,
+ 'type': 'Duration',
+ },
+ 'traffic_options': {
+ 'flow': 'ipv4_1flow_Packets_vpe.yaml',
+ 'imix': 'imix_voice.yaml',
+ },
+ 'vnf_options': {
+ 'vpe': {
+ 'cfg': 'vpe_config',
+ },
+ },
+ 'traffic_profile': 'ipv4_throughput_vpe.yaml',
+ 'type': 'NSPerf',
+ 'nodes': {
+ 'tg__1': 'trafficgen_1.yardstick',
+ 'vnf__1': 'vnf.yardstick',
+ },
+ 'topology': 'vpe_vnf_topology.yaml',
+ },
+ ],
+ 'context': {
+ 'nfvi_type': 'baremetal',
+ 'type': 'Node',
+ 'name': 'yardstick',
+ 'file': '/etc/yardstick/nodes/pod.yaml',
+ },
+ 'schema': 'yardstick:task:0.1',
+ }
+
+ @mock.patch(SSH_HELPER)
+ def test___init__(self, ssh):
+ mock_ssh(ssh)
+ trex_traffic_gen = TrexTrafficGenRFC('vnf1', self.VNFD_0)
+ self.assertIsNotNone(trex_traffic_gen.resource_helper._terminated.value)
+
+ @mock.patch(SSH_HELPER)
+ def test_collect_kpi(self, ssh):
+ mock_ssh(ssh)
+ trex_traffic_gen = TrexTrafficGenRFC('vnf1', self.VNFD_0)
+ self.assertEqual(trex_traffic_gen.collect_kpi(), {})
+
+ @mock.patch(SSH_HELPER)
+ def test_listen_traffic(self, ssh):
+ mock_ssh(ssh)
+ trex_traffic_gen = TrexTrafficGenRFC('vnf1', self.VNFD_0)
+ self.assertIsNone(trex_traffic_gen.listen_traffic({}))
+
+ @mock.patch(SSH_HELPER)
+ def test_instantiate(self, ssh):
+ mock_ssh(ssh)
+