from yardstick.network_services.nfvi.resource import ResourceProfile
from yardstick.network_services.traffic_profile.base import TrafficProfile
from yardstick.network_services.vnf_generic.vnf.base import VnfdHelper
-from yardstick.ssh import SSHError
class MockError(BaseException):
with self.assertRaises(NotImplementedError):
setup_env_helper.build_config()
- def test__get_ports_gateway(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- setup_env_helper = SetupEnvHelper(vnfd_helper, mock.Mock(), mock.Mock())
- result = setup_env_helper._get_ports_gateway("xe0")
- self.assertEqual(result, "152.16.100.20")
-
- result = setup_env_helper._get_ports_gateway("xe123")
- self.assertIsNone(result)
-
def test_setup_vnf_environment(self):
setup_env_helper = SetupEnvHelper(mock.Mock(), mock.Mock(), mock.Mock())
self.assertIsNone(setup_env_helper.setup_vnf_environment())
dpdk_setup_helper._build_pipeline_kwargs()
self.assertDictEqual(dpdk_setup_helper.pipeline_kwargs, expected)
- def test__get_app_cpu(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- ssh_helper.provision_tool.return_value = 'tool_path'
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
-
- dpdk_setup_helper.CORES = expected = [5, 4, 3]
- result = dpdk_setup_helper._get_app_cpu()
- self.assertEqual(result, expected)
-
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.CpuSysCores')
- def test__get_app_cpu_no_cores_sw(self, mock_cpu_sys_cores_class):
- mock_cpu_sys_cores = mock_cpu_sys_cores_class()
- mock_cpu_sys_cores.get_core_socket.return_value = {
- 'socket': [2, 4, 8, 10, 12],
- }
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- ssh_helper.provision_tool.return_value = 'tool_path'
- scenario_helper = mock.Mock()
- scenario_helper.vnf_cfg = {
- 'worker_threads': '2',
- }
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper.CORES = []
- dpdk_setup_helper.SW_DEFAULT_CORE = 1
- dpdk_setup_helper.HW_DEFAULT_CORE = 2
- dpdk_setup_helper.socket = 'socket'
-
- expected = [2, 4, 8]
- result = dpdk_setup_helper._get_app_cpu()
- self.assertEqual(result, expected)
-
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.CpuSysCores')
- def test__get_app_cpu_no_cores_hw(self, mock_cpu_sys_cores_class):
- mock_cpu_sys_cores = mock_cpu_sys_cores_class()
- mock_cpu_sys_cores.get_core_socket.return_value = {
- 'socket': [2, 4, 8, 10, 12],
- }
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- scenario_helper.vnf_cfg = {
- 'worker_threads': '2',
- 'lb_config': 'HW',
- }
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper.CORES = []
- dpdk_setup_helper.SW_DEFAULT_CORE = 1
- dpdk_setup_helper.HW_DEFAULT_CORE = 2
- dpdk_setup_helper.socket = 'socket'
-
- expected = [2, 4, 8, 10]
- result = dpdk_setup_helper._get_app_cpu()
- self.assertEqual(result, expected)
-
- def test__get_cpu_sibling_list(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- ssh_helper.execute.side_effect = iter([(0, '5', ''), (0, '3,4', ''), (0, '10', '')])
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper._get_app_cpu = mock.Mock(return_value=[])
-
- expected = ['5', '3', '4', '10']
- result = dpdk_setup_helper._get_cpu_sibling_list([1, 3, 7])
- self.assertEqual(result, expected)
-
- def test__get_cpu_sibling_list_no_core_arg(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- ssh_helper.execute.side_effect = iter([(0, '5', ''), (0, '3,4', ''), (0, '10', '')])
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper._get_app_cpu = mock.Mock(return_value=[1, 7])
-
- expected = ['5', '3', '4']
- result = dpdk_setup_helper._get_cpu_sibling_list()
- self.assertEqual(result, expected)
-
- def test__get_cpu_sibling_list_ssh_failure(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- ssh_helper.execute.side_effect = iter([(0, '5', ''), SSHError, (0, '10', '')])
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper._get_app_cpu = mock.Mock(return_value=[])
-
- expected = []
- result = dpdk_setup_helper._get_cpu_sibling_list([1, 3, 7])
- self.assertEqual(result, expected)
-
- def test__validate_cpu_cfg(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- ssh_helper.execute.side_effect = iter([(0, '5', ''), (0, '3,4', ''), (0, '10', '')])
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper._get_app_cpu = mock.Mock(return_value=[1, 3, 7])
-
- expected = ['5', '3', '4', '10']
- result = dpdk_setup_helper._validate_cpu_cfg()
- self.assertEqual(result, expected)
-
@mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
@mock.patch('yardstick.ssh.SSH')
def test_setup_vnf_environment(self, _, mock_time):
class TestSampleVNFDeployHelper(unittest.TestCase):
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
@mock.patch('subprocess.check_output')
- def test_deploy_vnfs_disabled(self, mock_check_output):
+ def test_deploy_vnfs_disabled(self, mock_check_output, mock_time):
vnfd_helper = mock.Mock()
ssh_helper = mock.Mock()
ssh_helper.join_bin_path.return_value = 'joined_path'
sample_vnf_deploy_helper = SampleVNFDeployHelper(vnfd_helper, ssh_helper)
self.assertIsNone(sample_vnf_deploy_helper.deploy_vnfs('name1'))
- self.assertEqual(ssh_helper.execute.call_count, 0)
- self.assertEqual(ssh_helper.put.call_count, 0)
+ sample_vnf_deploy_helper.DISABLE_DEPLOY = True
+ self.assertEqual(ssh_helper.execute.call_count, 5)
+ self.assertEqual(ssh_helper.put.call_count, 1)
@mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
@mock.patch('subprocess.check_output')
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
sample_vnf = SampleVNF('vnf1', vnfd)
sample_vnf.APP_NAME = 'sample1'
+ sample_vnf.WAIT_TIME_FOR_SCRIPT = 0
sample_vnf._start_server = mock.Mock(return_value=0)
sample_vnf._vnf_process = mock.MagicMock()
sample_vnf._vnf_process.exitcode = 0
sample_vnf_tg.terminate()
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.LOG')
- def test_wait_for_instantiate(self, mock_logger, mock_time):
+ def test__wait_for_process(self):
sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
- sample_vnf_tg._check_status = mock.Mock(side_effect=iter([1, 0]))
- sample_vnf_tg._tg_process = mock.Mock()
- sample_vnf_tg._tg_process.is_alive.return_value = True
- sample_vnf_tg._tg_process.exitcode = 234
-
- self.assertEqual(sample_vnf_tg.wait_for_instantiate(), 234)
-
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.LOG')
- def test_wait_for_instantiate_not_alive(self, mock_logger, mock_time):
+ with mock.patch.object(sample_vnf_tg, '_check_status',
+ return_value=0) as mock_status, \
+ mock.patch.object(sample_vnf_tg, '_tg_process') as mock_proc:
+ mock_proc.is_alive.return_value = True
+ mock_proc.exitcode = 234
+ self.assertEqual(sample_vnf_tg._wait_for_process(), 234)
+ mock_proc.is_alive.assert_called_once()
+ mock_status.assert_called_once()
+
+ def test__wait_for_process_not_alive(self):
sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
- sample_vnf_tg._check_status = mock.Mock(return_value=1)
- sample_vnf_tg._tg_process = mock.Mock()
- sample_vnf_tg._tg_process.is_alive.side_effect = iter([True, False])
- sample_vnf_tg._tg_process.exitcode = 234
-
- with self.assertRaises(RuntimeError):
- sample_vnf_tg.wait_for_instantiate()
-
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.LOG')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.Process')
- def test_wait_for_instantiate_delayed(self, mock_process, mock_logger, mock_time):
- class MockClientStarted(mock.Mock):
-
- def __init__(self, *args, **kwargs):
- super(MockClientStarted, self).__init__(*args, **kwargs)
- self.iter = iter([0, 0, 1])
-
- @property
- def value(self):
- return next(self.iter)
-
- mock_traffic_profile = mock.Mock(autospec=TrafficProfile)
- mock_traffic_profile.get_traffic_definition.return_value = "64"
- mock_traffic_profile.execute_traffic.return_value = "64"
- mock_traffic_profile.params = self.TRAFFIC_PROFILE
+ with mock.patch.object(sample_vnf_tg, '_tg_process') as mock_proc:
+ mock_proc.is_alive.return_value = False
+ self.assertRaises(RuntimeError, sample_vnf_tg._wait_for_process)
+ mock_proc.is_alive.assert_called_once()
+ def test__wait_for_process_delayed(self):
sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
- sample_vnf_tg._check_status = mock.Mock(side_effect=iter([1, 0]))
- sample_vnf_tg._tg_process = mock.Mock()
- sample_vnf_tg._tg_process.is_alive.return_value = True
- sample_vnf_tg._tg_process.exitcode = 234
- sample_vnf_tg.resource_helper = mock.Mock()
- sample_vnf_tg.resource_helper.client_started = MockClientStarted()
-
- self.assertTrue(sample_vnf_tg.run_traffic(mock_traffic_profile))
- self.assertEqual(mock_time.sleep.call_count, 2)
+ with mock.patch.object(sample_vnf_tg, '_check_status',
+ side_effect=[1, 0]) as mock_status, \
+ mock.patch.object(sample_vnf_tg,
+ '_tg_process') as mock_proc:
+ mock_proc.is_alive.return_value = True
+ mock_proc.exitcode = 234
+ self.assertEqual(sample_vnf_tg._wait_for_process(), 234)
+ mock_proc.is_alive.assert_has_calls([mock.call(), mock.call()])
+ mock_status.assert_has_calls([mock.call(), mock.call()])
Code Review / yardstick.git / blobdiff