aboutsummaryrefslogtreecommitdiffstats
path: root/yardstick/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py
diff options
context:
space:
mode:
Diffstat (limited to 'yardstick/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py')
-rw-r--r--yardstick/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py999
1 files changed, 405 insertions, 594 deletions
diff --git a/yardstick/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py
index 43682dd07..8aa59ccd1 100644
--- a/yardstick/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py
+++ b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2017-2018 Intel Corporation
+# Copyright (c) 2017-2019 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@@ -16,24 +16,25 @@ from copy import deepcopy
import unittest
import mock
+import six
+import time
+import subprocess
+
+import paramiko
from yardstick.common import exceptions as y_exceptions
from yardstick.common import utils
-from yardstick.network_services.nfvi.resource import ResourceProfile
-from yardstick.network_services.vnf_generic.vnf.base import VnfdHelper
+from yardstick.network_services.nfvi import resource
+from yardstick.network_services.vnf_generic.vnf import base
from yardstick.network_services.vnf_generic.vnf import sample_vnf
-from yardstick.network_services.vnf_generic.vnf.vnf_ssh_helper import VnfSshHelper
-from yardstick.network_services.vnf_generic.vnf.sample_vnf import SampleVNFDeployHelper
-from yardstick.network_services.vnf_generic.vnf.sample_vnf import ScenarioHelper
+from yardstick.network_services.vnf_generic.vnf import vnf_ssh_helper
from yardstick.network_services.vnf_generic.vnf.sample_vnf import ResourceHelper
-from yardstick.network_services.vnf_generic.vnf.sample_vnf import ClientResourceHelper
-from yardstick.network_services.vnf_generic.vnf.sample_vnf import Rfc2544ResourceHelper
from yardstick.network_services.vnf_generic.vnf.sample_vnf import SetupEnvHelper
from yardstick.network_services.vnf_generic.vnf.sample_vnf import SampleVNF
from yardstick.network_services.vnf_generic.vnf.sample_vnf import SampleVNFTrafficGen
-from yardstick.network_services.vnf_generic.vnf.sample_vnf import DpdkVnfSetupEnvHelper
from yardstick.tests.unit.network_services.vnf_generic.vnf import test_base
from yardstick.benchmark.contexts import base as ctx_base
+from yardstick import ssh
class MockError(Exception):
@@ -82,10 +83,11 @@ class TestVnfSshHelper(unittest.TestCase):
'virtual-interface': {
'dst_mac': '00:00:00:00:00:03',
'vpci': '0000:05:00.0',
+ 'dpdk_port_num': 0,
+ 'driver': 'i40e',
'local_ip': '152.16.100.19',
'type': 'PCI-PASSTHROUGH',
'netmask': '255.255.255.0',
- 'dpdk_port_num': 0,
'bandwidth': '10 Gbps',
'dst_ip': '152.16.100.20',
'local_mac': '00:00:00:00:00:01',
@@ -99,10 +101,11 @@ class TestVnfSshHelper(unittest.TestCase):
'virtual-interface': {
'dst_mac': '00:00:00:00:00:04',
'vpci': '0000:05:00.1',
+ 'dpdk_port_num': 1,
+ 'driver': 'ixgbe',
'local_ip': '152.16.40.19',
'type': 'PCI-PASSTHROUGH',
'netmask': '255.255.255.0',
- 'dpdk_port_num': 1,
'bandwidth': '10 Gbps',
'dst_ip': '152.16.40.20',
'local_mac': '00:00:00:00:00:02',
@@ -151,90 +154,88 @@ class TestVnfSshHelper(unittest.TestCase):
}
}
+ def setUp(self):
+ self.ssh_helper = vnf_ssh_helper.VnfSshHelper(
+ self.VNFD_0['mgmt-interface'], 'my/bin/path')
+ self.ssh_helper._run = mock.Mock()
+
def assertAll(self, iterable, message=None):
self.assertTrue(all(iterable), message)
def test_get_class(self):
- self.assertIs(VnfSshHelper.get_class(), VnfSshHelper)
+ self.assertIs(vnf_ssh_helper.VnfSshHelper.get_class(),
+ vnf_ssh_helper.VnfSshHelper)
- @mock.patch('yardstick.ssh.paramiko')
+ @mock.patch.object(ssh, 'paramiko')
def test_copy(self, _):
- ssh_helper = VnfSshHelper(self.VNFD_0['mgmt-interface'], 'my/bin/path')
- ssh_helper._run = mock.Mock()
-
- ssh_helper.execute('ls')
- self.assertTrue(ssh_helper.is_connected)
- result = ssh_helper.copy()
- self.assertIsInstance(result, VnfSshHelper)
+ self.ssh_helper.execute('ls')
+ self.assertTrue(self.ssh_helper.is_connected)
+ result = self.ssh_helper.copy()
+ self.assertIsInstance(result, vnf_ssh_helper.VnfSshHelper)
self.assertFalse(result.is_connected)
- self.assertEqual(result.bin_path, ssh_helper.bin_path)
- self.assertEqual(result.host, ssh_helper.host)
- self.assertEqual(result.port, ssh_helper.port)
- self.assertEqual(result.user, ssh_helper.user)
- self.assertEqual(result.password, ssh_helper.password)
- self.assertEqual(result.key_filename, ssh_helper.key_filename)
-
- @mock.patch('yardstick.ssh.paramiko')
+ self.assertEqual(result.bin_path, self.ssh_helper.bin_path)
+ self.assertEqual(result.host, self.ssh_helper.host)
+ self.assertEqual(result.port, self.ssh_helper.port)
+ self.assertEqual(result.user, self.ssh_helper.user)
+ self.assertEqual(result.password, self.ssh_helper.password)
+ self.assertEqual(result.key_filename, self.ssh_helper.key_filename)
+
+ @mock.patch.object(paramiko, 'SSHClient')
def test_upload_config_file(self, mock_paramiko):
- ssh_helper = VnfSshHelper(self.VNFD_0['mgmt-interface'], 'my/bin/path')
- ssh_helper._run = mock.MagicMock()
+ self.assertFalse(self.ssh_helper.is_connected)
+ cfg_file = self.ssh_helper.upload_config_file('/my/prefix', 'my content')
+ self.assertTrue(self.ssh_helper.is_connected)
+ mock_paramiko.assert_called_once()
+ self.assertEqual(cfg_file, '/my/prefix')
- self.assertFalse(ssh_helper.is_connected)
- cfg_file = ssh_helper.upload_config_file('my/prefix', 'my content')
- self.assertTrue(ssh_helper.is_connected)
- mock_paramiko.SSHClient.assert_called_once()
+ @mock.patch.object(paramiko, 'SSHClient')
+ def test_upload_config_file_path_does_not_exist(self, mock_paramiko):
+ self.assertFalse(self.ssh_helper.is_connected)
+ cfg_file = self.ssh_helper.upload_config_file('my/prefix', 'my content')
+ self.assertTrue(self.ssh_helper.is_connected)
+ mock_paramiko.assert_called_once()
self.assertTrue(cfg_file.startswith('/tmp'))
- cfg_file = ssh_helper.upload_config_file('/my/prefix', 'my content')
- self.assertTrue(ssh_helper.is_connected)
- mock_paramiko.SSHClient.assert_called_once()
- self.assertEqual(cfg_file, '/my/prefix')
-
def test_join_bin_path(self):
- ssh_helper = VnfSshHelper(self.VNFD_0['mgmt-interface'], 'my/bin/path')
-
expected_start = 'my'
expected_middle_list = ['bin']
expected_end = 'path'
- result = ssh_helper.join_bin_path()
+ result = self.ssh_helper.join_bin_path()
self.assertTrue(result.startswith(expected_start))
self.assertAll(middle in result for middle in expected_middle_list)
self.assertTrue(result.endswith(expected_end))
expected_middle_list.append(expected_end)
expected_end = 'some_file.sh'
- result = ssh_helper.join_bin_path('some_file.sh')
+ result = self.ssh_helper.join_bin_path('some_file.sh')
self.assertTrue(result.startswith(expected_start))
self.assertAll(middle in result for middle in expected_middle_list)
self.assertTrue(result.endswith(expected_end))
expected_middle_list.append('some_dir')
expected_end = 'some_file.sh'
- result = ssh_helper.join_bin_path('some_dir', 'some_file.sh')
+ result = self.ssh_helper.join_bin_path('some_dir', 'some_file.sh')
self.assertTrue(result.startswith(expected_start))
self.assertAll(middle in result for middle in expected_middle_list)
self.assertTrue(result.endswith(expected_end))
- @mock.patch('yardstick.ssh.paramiko')
- @mock.patch('yardstick.ssh.provision_tool')
+ @mock.patch.object(paramiko, 'SSHClient')
+ @mock.patch.object(ssh, 'provision_tool')
def test_provision_tool(self, mock_provision_tool, mock_paramiko):
- ssh_helper = VnfSshHelper(self.VNFD_0['mgmt-interface'], 'my/bin/path')
- ssh_helper._run = mock.MagicMock()
-
- self.assertFalse(ssh_helper.is_connected)
- ssh_helper.provision_tool()
- self.assertTrue(ssh_helper.is_connected)
- mock_paramiko.SSHClient.assert_called_once()
+ self.assertFalse(self.ssh_helper.is_connected)
+ self.ssh_helper.provision_tool()
+ self.assertTrue(self.ssh_helper.is_connected)
+ mock_paramiko.assert_called_once()
mock_provision_tool.assert_called_once()
- ssh_helper.provision_tool(tool_file='my_tool.sh')
- self.assertTrue(ssh_helper.is_connected)
- mock_paramiko.SSHClient.assert_called_once()
+ self.ssh_helper.provision_tool(tool_file='my_tool.sh')
+ self.assertTrue(self.ssh_helper.is_connected)
+ mock_paramiko.assert_called_once()
self.assertEqual(mock_provision_tool.call_count, 2)
- ssh_helper.provision_tool('tool_path', 'my_tool.sh')
- self.assertTrue(ssh_helper.is_connected)
- mock_paramiko.SSHClient.assert_called_once()
+ self.ssh_helper.provision_tool('tool_path', 'my_tool.sh')
+ self.assertTrue(self.ssh_helper.is_connected)
+ mock_paramiko.assert_called_once()
self.assertEqual(mock_provision_tool.call_count, 3)
@@ -360,123 +361,24 @@ class TestSetupEnvHelper(unittest.TestCase):
class TestDpdkVnfSetupEnvHelper(unittest.TestCase):
- VNFD_0 = {
- 'short-name': 'VpeVnf',
- 'vdu': [
- {
- 'routing_table': [
- {
- 'network': '152.16.100.20',
- 'netmask': '255.255.255.0',
- 'gateway': '152.16.100.20',
- 'if': 'xe0'
- },
- {
- 'network': '152.16.40.20',
- 'netmask': '255.255.255.0',
- 'gateway': '152.16.40.20',
- 'if': 'xe1'
- },
- ],
- 'description': 'VPE approximation using DPDK',
- 'name': 'vpevnf-baremetal',
- 'nd_route_tbl': [
- {
- 'network': '0064:ff9b:0:0:0:0:9810:6414',
- 'netmask': '112',
- 'gateway': '0064:ff9b:0:0:0:0:9810:6414',
- 'if': 'xe0'
- },
- {
- 'network': '0064:ff9b:0:0:0:0:9810:2814',
- 'netmask': '112',
- 'gateway': '0064:ff9b:0:0:0:0:9810:2814',
- 'if': 'xe1'
- },
- ],
- 'id': 'vpevnf-baremetal',
- 'external-interface': [
- {
- 'virtual-interface': {
- 'dst_mac': '00:00:00:00:00:03',
- 'vpci': '0000:05:00.0',
- 'dpdk_port_num': 0,
- 'driver': 'i40e',
- 'local_ip': '152.16.100.19',
- 'type': 'PCI-PASSTHROUGH',
- 'netmask': '255.255.255.0',
- 'bandwidth': '10 Gbps',
- 'dst_ip': '152.16.100.20',
- 'local_mac': '00:00:00:00:00:01',
- 'vld_id': 'uplink_0',
- 'ifname': 'xe0',
- },
- 'vnfd-connection-point-ref': 'xe0',
- 'name': 'xe0'
- },
- {
- 'virtual-interface': {
- 'dst_mac': '00:00:00:00:00:04',
- 'vpci': '0000:05:00.1',
- 'dpdk_port_num': 1,
- 'driver': 'ixgbe',
- 'local_ip': '152.16.40.19',
- 'type': 'PCI-PASSTHROUGH',
- 'netmask': '255.255.255.0',
- 'bandwidth': '10 Gbps',
- 'dst_ip': '152.16.40.20',
- 'local_mac': '00:00:00:00:00:02',
- 'vld_id': 'downlink_0',
- 'ifname': 'xe1',
- },
- 'vnfd-connection-point-ref': 'xe1',
- 'name': 'xe1'
- },
- ],
- },
- ],
- 'description': 'Vpe approximation using DPDK',
- 'mgmt-interface': {
- 'vdu-id': 'vpevnf-baremetal',
- 'host': '1.1.1.1',
- 'password': 'r00t',
- 'user': 'root',
- 'ip': '1.1.1.1'
- },
- 'benchmark': {
- 'kpi': [
- 'packets_in',
- 'packets_fwd',
- 'packets_dropped',
- ],
- },
- 'connection-point': [
- {
- 'type': 'VPORT',
- 'name': 'xe0',
- },
- {
- 'type': 'VPORT',
- 'name': 'xe1',
- },
- ],
- 'id': 'VpeApproxVnf', 'name': 'VPEVnfSsh'
- }
+ VNFD_0 = TestVnfSshHelper.VNFD_0
- VNFD = {
- 'vnfd:vnfd-catalog': {
- 'vnfd': [
- VNFD_0,
- ]
- }
- }
+ VNFD = TestVnfSshHelper.VNFD
+
+ def setUp(self):
+ self.vnfd_helper = base.VnfdHelper(deepcopy(self.VNFD_0))
+ self.scenario_helper = mock.Mock()
+ self.ssh_helper = mock.Mock()
+ self.dpdk_setup_helper = sample_vnf.DpdkVnfSetupEnvHelper(
+ self.vnfd_helper, self.ssh_helper, self.scenario_helper)
def test__update_packet_type(self):
ip_pipeline_cfg = 'pkt_type = ipv4'
pkt_type = {'pkt_type': '1'}
expected = "pkt_type = 1"
- result = DpdkVnfSetupEnvHelper._update_packet_type(ip_pipeline_cfg, pkt_type)
+ result = self.dpdk_setup_helper._update_packet_type(
+ ip_pipeline_cfg, pkt_type)
self.assertEqual(result, expected)
def test__update_packet_type_no_op(self):
@@ -484,91 +386,99 @@ class TestDpdkVnfSetupEnvHelper(unittest.TestCase):
pkt_type = {'pkt_type': '1'}
expected = "pkt_type = ipv6"
- result = DpdkVnfSetupEnvHelper._update_packet_type(ip_pipeline_cfg, pkt_type)
+ result = self.dpdk_setup_helper._update_packet_type(
+ ip_pipeline_cfg, pkt_type)
self.assertEqual(result, expected)
def test__update_packet_type_multi_op(self):
ip_pipeline_cfg = 'pkt_type = ipv4\npkt_type = 1\npkt_type = ipv4'
pkt_type = {'pkt_type': '1'}
-
expected = 'pkt_type = 1\npkt_type = 1\npkt_type = 1'
- result = DpdkVnfSetupEnvHelper._update_packet_type(ip_pipeline_cfg, pkt_type)
+
+ result = self.dpdk_setup_helper._update_packet_type(
+ ip_pipeline_cfg, pkt_type)
self.assertEqual(result, expected)
def test__update_traffic_type(self):
ip_pipeline_cfg = 'pkt_type = ipv4'
-
- traffic_options = {"vnf_type": DpdkVnfSetupEnvHelper.APP_NAME, 'traffic_type': 4}
+ traffic_options = {
+ "vnf_type": sample_vnf.DpdkVnfSetupEnvHelper.APP_NAME,
+ "traffic_type": 4}
expected = "pkt_type = ipv4"
- result = DpdkVnfSetupEnvHelper._update_traffic_type(ip_pipeline_cfg, traffic_options)
+
+ result = self.dpdk_setup_helper._update_traffic_type(
+ ip_pipeline_cfg, traffic_options)
self.assertEqual(result, expected)
def test__update_traffic_type_ipv6(self):
ip_pipeline_cfg = 'pkt_type = ipv4'
-
- traffic_options = {"vnf_type": DpdkVnfSetupEnvHelper.APP_NAME, 'traffic_type': 6}
+ traffic_options = {
+ "vnf_type": sample_vnf.DpdkVnfSetupEnvHelper.APP_NAME,
+ "traffic_type": 6}
expected = "pkt_type = ipv6"
- result = DpdkVnfSetupEnvHelper._update_traffic_type(ip_pipeline_cfg, traffic_options)
+
+ result = self.dpdk_setup_helper._update_traffic_type(
+ ip_pipeline_cfg, traffic_options)
self.assertEqual(result, expected)
def test__update_traffic_type_not_app_name(self):
ip_pipeline_cfg = 'traffic_type = 4'
-
- vnf_type = ''.join(["Not", DpdkVnfSetupEnvHelper.APP_NAME])
+ vnf_type = ''.join(["Not", sample_vnf.DpdkVnfSetupEnvHelper.APP_NAME])
traffic_options = {"vnf_type": vnf_type, 'traffic_type': 8}
expected = "traffic_type = 8"
- result = DpdkVnfSetupEnvHelper._update_traffic_type(ip_pipeline_cfg, traffic_options)
+
+ result = self.dpdk_setup_helper._update_traffic_type(
+ ip_pipeline_cfg, traffic_options)
self.assertEqual(result, expected)
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.open')
+ @mock.patch.object(six.moves.builtins, 'open')
@mock.patch.object(utils, 'find_relative_file')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.MultiPortConfig')
- @mock.patch.object(utils, 'open_relative_file')
- def test_build_config(self, mock_open_rf, mock_multi_port_config_class, mock_find, *args):
+ @mock.patch.object(sample_vnf, 'MultiPortConfig')
+ def test_build_config(self, mock_multi_port_config_class,
+ mock_find, *args):
mock_multi_port_config = mock_multi_port_config_class()
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- scenario_helper.vnf_cfg = {}
- scenario_helper.options = {}
- scenario_helper.all_options = {}
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
+ self.scenario_helper.vnf_cfg = {}
+ self.scenario_helper.options = {}
+ self.scenario_helper.all_options = {}
- dpdk_setup_helper.PIPELINE_COMMAND = expected = 'pipeline command'
- result = dpdk_setup_helper.build_config()
+ self.dpdk_setup_helper.PIPELINE_COMMAND = expected = 'pipeline command'
+ result = self.dpdk_setup_helper.build_config()
self.assertEqual(result, expected)
- self.assertGreaterEqual(ssh_helper.upload_config_file.call_count, 2)
+ self.assertGreaterEqual(self.ssh_helper.upload_config_file.call_count, 2)
mock_find.assert_called()
mock_multi_port_config.generate_config.assert_called()
mock_multi_port_config.generate_script.assert_called()
- scenario_helper.options = {'rules': 'fake_file'}
- scenario_helper.vnf_cfg = {'file': 'fake_file'}
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
+ @mock.patch.object(six.moves.builtins, 'open')
+ @mock.patch.object(utils, 'find_relative_file')
+ @mock.patch.object(sample_vnf, 'MultiPortConfig')
+ @mock.patch.object(utils, 'open_relative_file')
+ def test_build_config2(self, mock_open_rf, mock_multi_port_config_class,
+ mock_find, *args):
+ mock_multi_port_config = mock_multi_port_config_class()
+ self.scenario_helper.options = {'rules': 'fake_file'}
+ self.scenario_helper.vnf_cfg = {'file': 'fake_file'}
+ self.scenario_helper.all_options = {}
mock_open_rf.side_effect = mock.mock_open(read_data='fake_data')
- dpdk_setup_helper.PIPELINE_COMMAND = expected = 'pipeline command'
+ self.dpdk_setup_helper.PIPELINE_COMMAND = expected = 'pipeline command'
- result = dpdk_setup_helper.build_config()
+ result = self.dpdk_setup_helper.build_config()
mock_open_rf.assert_called()
self.assertEqual(result, expected)
- self.assertGreaterEqual(ssh_helper.upload_config_file.call_count, 2)
+ self.assertGreaterEqual(self.ssh_helper.upload_config_file.call_count, 2)
mock_find.assert_called()
mock_multi_port_config.generate_config.assert_called()
mock_multi_port_config.generate_script.assert_called()
def test__build_pipeline_kwargs(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.CFG_CONFIG = 'config'
- dpdk_setup_helper.CFG_SCRIPT = 'script'
- dpdk_setup_helper.pipeline_kwargs = {}
- dpdk_setup_helper.all_ports = [0, 1, 2]
- dpdk_setup_helper.scenario_helper.vnf_cfg = {'lb_config': 'HW',
- 'worker_threads': 1}
+ self.ssh_helper.provision_tool.return_value = 'tool_path'
+ self.dpdk_setup_helper.CFG_CONFIG = 'config'
+ self.dpdk_setup_helper.CFG_SCRIPT = 'script'
+ self.dpdk_setup_helper.pipeline_kwargs = {}
+ self.dpdk_setup_helper.all_ports = [0, 1, 2]
+ self.dpdk_setup_helper.scenario_helper.vnf_cfg = {'lb_config': 'HW',
+ 'worker_threads': 1}
expected = {
'cfg_file': 'config',
@@ -577,12 +487,14 @@ class TestDpdkVnfSetupEnvHelper(unittest.TestCase):
'tool_path': 'tool_path',
'hwlb': ' --hwlb 1',
}
- dpdk_setup_helper._build_pipeline_kwargs()
- self.assertDictEqual(dpdk_setup_helper.pipeline_kwargs, expected)
+ self.dpdk_setup_helper._build_pipeline_kwargs()
+ self.assertDictEqual(self.dpdk_setup_helper.pipeline_kwargs, expected)
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
- @mock.patch('yardstick.ssh.SSH')
+ @mock.patch.object(time, 'sleep')
+ @mock.patch.object(ssh, 'SSH')
def test_setup_vnf_environment(self, *args):
+ self.scenario_helper.nodes = [None, None]
+
def execute(cmd):
if cmd.startswith('which '):
return exec_failure
@@ -590,104 +502,82 @@ class TestDpdkVnfSetupEnvHelper(unittest.TestCase):
exec_success = (0, 'good output', '')
exec_failure = (1, 'bad output', 'error output')
+ self.ssh_helper.execute = execute
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- ssh_helper.execute = execute
+ self.dpdk_setup_helper._validate_cpu_cfg = mock.Mock(return_value=[])
- scenario_helper = mock.Mock()
- scenario_helper.nodes = [None, None]
- dpdk_vnf_setup_env_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_vnf_setup_env_helper._validate_cpu_cfg = mock.Mock(return_value=[])
-
- with mock.patch.object(dpdk_vnf_setup_env_helper, '_setup_dpdk'):
+ with mock.patch.object(self.dpdk_setup_helper, '_setup_dpdk'):
self.assertIsInstance(
- dpdk_vnf_setup_env_helper.setup_vnf_environment(),
- ResourceProfile)
+ self.dpdk_setup_helper.setup_vnf_environment(),
+ resource.ResourceProfile)
@mock.patch.object(utils, 'setup_hugepages')
def test__setup_dpdk(self, mock_setup_hugepages):
- ssh_helper = mock.Mock()
- ssh_helper.execute = mock.Mock()
- ssh_helper.execute.return_value = (0, 0, 0)
- scenario_helper = mock.Mock()
- scenario_helper.all_options = {'hugepages_gb': 8}
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(mock.ANY, ssh_helper,
- scenario_helper)
- dpdk_setup_helper._setup_dpdk()
- mock_setup_hugepages.assert_called_once_with(ssh_helper, 8*1024*1024)
- ssh_helper.execute.assert_has_calls([
+ self.ssh_helper.execute = mock.Mock()
+ self.ssh_helper.execute.return_value = (0, 0, 0)
+ self.scenario_helper.all_options = {'hugepages_gb': 8}
+ self.dpdk_setup_helper._setup_dpdk()
+ mock_setup_hugepages.assert_called_once_with(
+ self.ssh_helper, 8*1024*1024)
+ self.ssh_helper.execute.assert_has_calls([
mock.call('sudo modprobe uio && sudo modprobe igb_uio'),
mock.call('lsmod | grep -i igb_uio')
])
- @mock.patch('yardstick.ssh.SSH')
+ @mock.patch.object(ssh, 'SSH')
def test__setup_resources(self, _):
- vnfd_helper = VnfdHelper(deepcopy(self.VNFD_0))
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper._validate_cpu_cfg = mock.Mock()
-
- dpdk_setup_helper.bound_pci = [v['virtual-interface']["vpci"] for v in
- vnfd_helper.interfaces]
- result = dpdk_setup_helper._setup_resources()
- self.assertIsInstance(result, ResourceProfile)
- self.assertEqual(dpdk_setup_helper.socket, 0)
-
- @mock.patch('yardstick.ssh.SSH')
+ self.dpdk_setup_helper._validate_cpu_cfg = mock.Mock()
+ self.dpdk_setup_helper.bound_pci = [v['virtual-interface']["vpci"] for v in
+ self.vnfd_helper.interfaces]
+ result = self.dpdk_setup_helper._setup_resources()
+ self.assertIsInstance(result, resource.ResourceProfile)
+ self.assertEqual(self.dpdk_setup_helper.socket, 0)
+
+ @mock.patch.object(ssh, 'SSH')
def test__setup_resources_socket_1(self, _):
- vnfd_helper = VnfdHelper(deepcopy(self.VNFD_0))
- vnfd_helper.interfaces[0]['virtual-interface']['vpci'] = '0000:55:00.0'
- vnfd_helper.interfaces[1]['virtual-interface']['vpci'] = '0000:35:00.0'
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper._validate_cpu_cfg = mock.Mock()
-
- dpdk_setup_helper.bound_pci = [v['virtual-interface']["vpci"] for v in
- vnfd_helper.interfaces]
- result = dpdk_setup_helper._setup_resources()
- self.assertIsInstance(result, ResourceProfile)
- self.assertEqual(dpdk_setup_helper.socket, 1)
-
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
+ self.vnfd_helper.interfaces[0]['virtual-interface']['vpci'] = \
+ '0000:55:00.0'
+ self.vnfd_helper.interfaces[1]['virtual-interface']['vpci'] = \
+ '0000:35:00.0'
+
+ self.dpdk_setup_helper._validate_cpu_cfg = mock.Mock()
+ self.dpdk_setup_helper.bound_pci = [v['virtual-interface']["vpci"] for v in
+ self.vnfd_helper.interfaces]
+ result = self.dpdk_setup_helper._setup_resources()
+ self.assertIsInstance(result, resource.ResourceProfile)
+ self.assertEqual(self.dpdk_setup_helper.socket, 1)
+
+ @mock.patch.object(time, 'sleep')
def test__detect_and_bind_drivers(self, *args):
- vnfd_helper = VnfdHelper(deepcopy(self.VNFD_0))
- ssh_helper = mock.Mock()
- # ssh_helper.execute = mock.Mock(return_value = (0, 'text', ''))
- # ssh_helper.execute.return_value = 0, 'output', ''
- scenario_helper = mock.Mock()
- scenario_helper.nodes = [None, None]
+ self.scenario_helper.nodes = [None, None]
rv = ['0000:05:00.1', '0000:05:00.0']
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper.dpdk_bind_helper._get_bound_pci_addresses = mock.Mock(return_value=rv)
- dpdk_setup_helper.dpdk_bind_helper.bind = mock.Mock()
- dpdk_setup_helper.dpdk_bind_helper.read_status = mock.Mock()
+ self.dpdk_setup_helper.dpdk_bind_helper._get_bound_pci_addresses = \
+ mock.Mock(return_value=rv)
+ self.dpdk_setup_helper.dpdk_bind_helper.bind = mock.Mock()
+ self.dpdk_setup_helper.dpdk_bind_helper.read_status = mock.Mock()
- self.assertIsNone(dpdk_setup_helper._detect_and_bind_drivers())
+ self.assertIsNone(self.dpdk_setup_helper._detect_and_bind_drivers())
- intf_0 = vnfd_helper.vdu[0]['external-interface'][0]['virtual-interface']
- intf_1 = vnfd_helper.vdu[0]['external-interface'][1]['virtual-interface']
+ intf_0 = self.vnfd_helper.vdu[0]['external-interface'][0]['virtual-interface']
+ intf_1 = self.vnfd_helper.vdu[0]['external-interface'][1]['virtual-interface']
self.assertEqual(0, intf_0['dpdk_port_num'])
self.assertEqual(1, intf_1['dpdk_port_num'])
def test_tear_down(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- scenario_helper.nodes = [None, None]
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper.dpdk_bind_helper.bind = mock.Mock()
- dpdk_setup_helper.dpdk_bind_helper.used_drivers = {
+ self.scenario_helper.nodes = [None, None]
+
+ self.dpdk_setup_helper.dpdk_bind_helper.bind = mock.Mock()
+ self.dpdk_setup_helper.dpdk_bind_helper.used_drivers = {
'd1': ['0000:05:00.0'],
'd3': ['0000:05:01.0'],
}
- self.assertIsNone(dpdk_setup_helper.tear_down())
- dpdk_setup_helper.dpdk_bind_helper.bind.assert_any_call(['0000:05:00.0'], 'd1', True)
- dpdk_setup_helper.dpdk_bind_helper.bind.assert_any_call(['0000:05:01.0'], 'd3', True)
+ self.assertIsNone(self.dpdk_setup_helper.tear_down())
+ self.dpdk_setup_helper.dpdk_bind_helper.bind.assert_any_call(
+ ['0000:05:00.0'], 'd1', True)
+ self.dpdk_setup_helper.dpdk_bind_helper.bind.assert_any_call(
+ ['0000:05:01.0'], 'd3', True)
class TestResourceHelper(unittest.TestCase):
@@ -791,46 +681,33 @@ class TestResourceHelper(unittest.TestCase):
'id': 'VpeApproxVnf', 'name': 'VPEVnfSsh'
}
+ def setUp(self):
+ self.vnfd_helper = base.VnfdHelper(self.VNFD_0)
+ self.dpdk_setup_helper = sample_vnf.DpdkVnfSetupEnvHelper(
+ self.vnfd_helper, mock.Mock(), mock.Mock())
+ self.resource_helper = sample_vnf.ResourceHelper(self.dpdk_setup_helper)
+
def test_setup(self):
resource = object()
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- dpdk_setup_helper.setup_vnf_environment = mock.Mock(return_value=resource)
- resource_helper = ResourceHelper(dpdk_setup_helper)
+ self.dpdk_setup_helper.setup_vnf_environment = (
+ mock.Mock(return_value=resource))
+ resource_helper = sample_vnf.ResourceHelper(self.dpdk_setup_helper)
self.assertIsNone(resource_helper.setup())
self.assertIs(resource_helper.resource, resource)
def test_generate_cfg(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- resource_helper = ResourceHelper(dpdk_setup_helper)
-
- self.assertIsNone(resource_helper.generate_cfg())
+ self.assertIsNone(self.resource_helper.generate_cfg())
def test_stop_collect(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- resource_helper = ResourceHelper(dpdk_setup_helper)
- resource_helper.resource = mock.Mock()
+ self.resource_helper.resource = mock.Mock()
- self.assertIsNone(resource_helper.stop_collect())
+ self.assertIsNone(self.resource_helper.stop_collect())
def test_stop_collect_none(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- resource_helper = ResourceHelper(dpdk_setup_helper)
- resource_helper.resource = None
+ self.resource_helper.resource = None
- self.assertIsNone(resource_helper.stop_collect())
+ self.assertIsNone(self.resource_helper.stop_collect())
class TestClientResourceHelper(unittest.TestCase):
@@ -962,154 +839,75 @@ class TestClientResourceHelper(unittest.TestCase):
},
}
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.LOG')
- @mock.patch.object(sample_vnf, 'STLError', new_callable=lambda: MockError)
- def test_get_stats_not_connected(self, mock_stl_error, *args):
- vnfd_helper = VnfdHelper(self.VNFD_0)
+ def setUp(self):
+ vnfd_helper = base.VnfdHelper(self.VNFD_0)
ssh_helper = mock.Mock()
scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(
+ dpdk_setup_helper = sample_vnf.DpdkVnfSetupEnvHelper(
vnfd_helper, ssh_helper, scenario_helper)
- client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
- client_resource_helper.client = mock.Mock()
- client_resource_helper.client.get_stats.side_effect = mock_stl_error
+ self.client_resource_helper = (
+ sample_vnf.ClientResourceHelper(dpdk_setup_helper))
- self.assertEqual(client_resource_helper.get_stats(), {})
- client_resource_helper.client.get_stats.assert_called_once()
+ @mock.patch.object(sample_vnf, 'LOG')
+ @mock.patch.object(sample_vnf, 'STLError', new_callable=lambda: MockError)
+ def test_get_stats_not_connected(self, mock_stl_error, *args):
+ self.client_resource_helper.client = mock.Mock()
+ self.client_resource_helper.client.get_stats.side_effect = \
+ mock_stl_error
+
+ self.assertEqual(self.client_resource_helper.get_stats(), {})
+ self.client_resource_helper.client.get_stats.assert_called_once()
def test_clear_stats(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(
- vnfd_helper, ssh_helper, scenario_helper)
- client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
- client_resource_helper.client = mock.Mock()
+ self.client_resource_helper.client = mock.Mock()
- self.assertIsNone(client_resource_helper.clear_stats())
+ self.assertIsNone(self.client_resource_helper.clear_stats())
self.assertEqual(
- client_resource_helper.client.clear_stats.call_count, 1)
+ self.client_resource_helper.client.clear_stats.call_count, 1)
def test_clear_stats_of_ports(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(
- vnfd_helper, ssh_helper, scenario_helper)
- client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
- client_resource_helper.client = mock.Mock()
+ self.client_resource_helper.client = mock.Mock()
- self.assertIsNone(client_resource_helper.clear_stats([3, 4]))
- self.assertEqual(
- client_resource_helper.client.clear_stats.call_count, 1)
+ self.assertIsNone(self.client_resource_helper.clear_stats([3, 4]))
+ self.client_resource_helper.client.clear_stats.assert_called_once()
def test_start(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(
- vnfd_helper, ssh_helper, scenario_helper)
- client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
- client_resource_helper.client = mock.Mock()
+ self.client_resource_helper.client = mock.Mock()
- self.assertIsNone(client_resource_helper.start())
- client_resource_helper.client.start.assert_called_once()
+ self.assertIsNone(self.client_resource_helper.start())
+ self.client_resource_helper.client.start.assert_called_once()
def test_start_ports(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(
- vnfd_helper, ssh_helper, scenario_helper)
- client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
- client_resource_helper.client = mock.Mock()
+ self.client_resource_helper.client = mock.Mock()
- self.assertIsNone(client_resource_helper.start([3, 4]))
- client_resource_helper.client.start.assert_called_once()
+ self.assertIsNone(self.client_resource_helper.start([3, 4]))
+ self.client_resource_helper.client.start.assert_called_once()
def test_collect_kpi_with_queue(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
- client_resource_helper._result = {'existing': 43, 'replaceable': 12}
- client_resource_helper._queue = mock.Mock()
- client_resource_helper._queue.empty.return_value = False
- client_resource_helper._queue.get.return_value = {'incoming': 34, 'replaceable': 99}
+ self.client_resource_helper._result = {
+ 'existing': 43,
+ 'replaceable': 12}
+ self.client_resource_helper._queue = mock.Mock()
+ self.client_resource_helper._queue.empty.return_value = False
+ self.client_resource_helper._queue.get.return_value = {
+ 'incoming': 34,
+ 'replaceable': 99}
expected = {
'existing': 43,
'incoming': 34,
'replaceable': 99,
}
- result = client_resource_helper.collect_kpi()
- self.assertDictEqual(result, expected)
+ result = self.client_resource_helper.collect_kpi()
+ self.assertEqual(result, expected)
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
+ @mock.patch.object(time, 'sleep')
@mock.patch.object(sample_vnf, 'STLError')
def test__connect_with_failures(self, mock_stl_error, *args):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- ssh_helper = mock.Mock()
- scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
client = mock.MagicMock()
client.connect.side_effect = mock_stl_error(msg='msg')
- self.assertIs(client_resource_helper._connect(client), client)
-
- @mock.patch.object(ClientResourceHelper, '_build_ports')
- @mock.patch.object(ClientResourceHelper, '_run_traffic_once',
- return_value=(True, mock.ANY))
- def test_run_traffic(self, mock_run_traffic_once, mock_build_ports):
- client_resource_helper = ClientResourceHelper(mock.Mock())
- client = mock.Mock()
- traffic_profile = mock.Mock()
- mq_producer = mock.Mock()
- with mock.patch.object(client_resource_helper, '_connect') \
- as mock_connect, \
- mock.patch.object(client_resource_helper, '_terminated') \
- as mock_terminated:
- mock_connect.return_value = client
- type(mock_terminated).value = mock.PropertyMock(
- side_effect=[0, 1, 1, lambda x: x])
- client_resource_helper.run_traffic(traffic_profile, mq_producer)
-
- mock_build_ports.assert_called_once()
- traffic_profile.register_generator.assert_called_once()
- mq_producer.tg_method_started.assert_called_once()
- mq_producer.tg_method_finished.assert_called_once()
- mq_producer.tg_method_iteration.assert_called_once_with(1)
- mock_run_traffic_once.assert_called_once_with(traffic_profile)
-
- @mock.patch.object(ClientResourceHelper, '_build_ports')
- @mock.patch.object(ClientResourceHelper, '_run_traffic_once',
- side_effect=Exception)
- def test_run_traffic_exception(self, mock_run_traffic_once,
- mock_build_ports):
- client_resource_helper = ClientResourceHelper(mock.Mock())
- client = mock.Mock()
- traffic_profile = mock.Mock()
- mq_producer = mock.Mock()
- with mock.patch.object(client_resource_helper, '_connect') \
- as mock_connect, \
- mock.patch.object(client_resource_helper, '_terminated') \
- as mock_terminated:
- mock_connect.return_value = client
- type(mock_terminated).value = mock.PropertyMock(return_value=0)
- mq_producer.reset_mock()
- # NOTE(ralonsoh): "trex_stl_exceptions.STLError" is mocked
- with self.assertRaises(Exception):
- client_resource_helper.run_traffic(traffic_profile,
- mq_producer)
-
- mock_build_ports.assert_called_once()
- traffic_profile.register_generator.assert_called_once()
- mock_run_traffic_once.assert_called_once_with(traffic_profile)
- mq_producer.tg_method_started.assert_called_once()
- mq_producer.tg_method_finished.assert_not_called()
- mq_producer.tg_method_iteration.assert_not_called()
+ self.assertIs(self.client_resource_helper._connect(client), client)
class TestRfc2544ResourceHelper(unittest.TestCase):
@@ -1156,185 +954,170 @@ class TestRfc2544ResourceHelper(unittest.TestCase):
}
}
- def test_property_rfc2544(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_1
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ def setUp(self):
+ self.scenario_helper = sample_vnf.ScenarioHelper('name1')
+ self.rfc2544_resource_helper = \
+ sample_vnf.Rfc2544ResourceHelper(self.scenario_helper)
- self.assertIsNone(rfc2544_resource_helper._rfc2544)
- self.assertDictEqual(rfc2544_resource_helper.rfc2544, self.RFC2544_CFG_1)
- self.assertDictEqual(rfc2544_resource_helper._rfc2544, self.RFC2544_CFG_1)
- scenario_helper.scenario_cfg = {} # ensure that resource_helper caches
- self.assertDictEqual(rfc2544_resource_helper.rfc2544, self.RFC2544_CFG_1)
+ def test_property_rfc2544(self):
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_1
+
+ self.assertIsNone(self.rfc2544_resource_helper._rfc2544)
+ self.assertEqual(self.rfc2544_resource_helper.rfc2544,
+ self.RFC2544_CFG_1)
+ self.assertEqual(self.rfc2544_resource_helper._rfc2544,
+ self.RFC2544_CFG_1)
+ # ensure that resource_helper caches
+ self.scenario_helper.scenario_cfg = {}
+ self.assertEqual(self.rfc2544_resource_helper.rfc2544,
+ self.RFC2544_CFG_1)
def test_property_tolerance_high(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_1
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_1
- self.assertIsNone(rfc2544_resource_helper._tolerance_high)
- self.assertEqual(rfc2544_resource_helper.tolerance_high, 0.15)
- self.assertEqual(rfc2544_resource_helper._tolerance_high, 0.15)
- self.assertEqual(rfc2544_resource_helper._tolerance_precision, 2)
- scenario_helper.scenario_cfg = {} # ensure that resource_helper caches
- self.assertEqual(rfc2544_resource_helper.tolerance_high, 0.15)
+ self.assertIsNone(self.rfc2544_resource_helper._tolerance_high)
+ self.assertEqual(self.rfc2544_resource_helper.tolerance_high, 0.15)
+ self.assertEqual(self.rfc2544_resource_helper._tolerance_high, 0.15)
+ self.assertEqual(self.rfc2544_resource_helper._tolerance_precision, 2)
+ # ensure that resource_helper caches
+ self.scenario_helper.scenario_cfg = {}
+ self.assertEqual(self.rfc2544_resource_helper.tolerance_high, 0.15)
def test_property_tolerance_low(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_1
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_1
- self.assertIsNone(rfc2544_resource_helper._tolerance_low)
- self.assertEqual(rfc2544_resource_helper.tolerance_low, 0.1)
- self.assertEqual(rfc2544_resource_helper._tolerance_low, 0.1)
- scenario_helper.scenario_cfg = {} # ensure that resource_helper caches
- self.assertEqual(rfc2544_resource_helper.tolerance_low, 0.1)
+ self.assertIsNone(self.rfc2544_resource_helper._tolerance_low)
+ self.assertEqual(self.rfc2544_resource_helper.tolerance_low, 0.1)
+ self.assertEqual(self.rfc2544_resource_helper._tolerance_low, 0.1)
+ # ensure that resource_helper caches
+ self.scenario_helper.scenario_cfg = {}
+ self.assertEqual(self.rfc2544_resource_helper.tolerance_low, 0.1)
def test_property_tolerance_high_range_swap(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_2
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_2
- self.assertEqual(rfc2544_resource_helper.tolerance_high, 0.25)
+ self.assertEqual(self.rfc2544_resource_helper.tolerance_high, 0.25)
def test_property_tolerance_low_range_swap(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_2
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_2
- self.assertEqual(rfc2544_resource_helper.tolerance_low, 0.05)
+ self.assertEqual(self.rfc2544_resource_helper.tolerance_low, 0.05)
def test_property_tolerance_high_not_range(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_3
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_3
- self.assertEqual(rfc2544_resource_helper.tolerance_high, 0.2)
- self.assertEqual(rfc2544_resource_helper._tolerance_precision, 1)
+ self.assertEqual(self.rfc2544_resource_helper.tolerance_high, 0.2)
+ self.assertEqual(self.rfc2544_resource_helper._tolerance_precision, 1)
def test_property_tolerance_low_not_range(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_3
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_3
- self.assertEqual(rfc2544_resource_helper.tolerance_low, 0.2)
+ self.assertEqual(self.rfc2544_resource_helper.tolerance_low, 0.2)
def test_property_tolerance_high_default(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_4
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_4
- self.assertEqual(rfc2544_resource_helper.tolerance_high, 0.0001)
+ self.assertEqual(self.rfc2544_resource_helper.tolerance_high, 0.0001)
def test_property_tolerance_low_default(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_4
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_4
- self.assertEqual(rfc2544_resource_helper.tolerance_low, 0.0001)
+ self.assertEqual(self.rfc2544_resource_helper.tolerance_low, 0.0001)
def test_property_latency(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_1
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_1
- self.assertIsNone(rfc2544_resource_helper._latency)
- self.assertTrue(rfc2544_resource_helper.latency)
- self.assertTrue(rfc2544_resource_helper._latency)
- scenario_helper.scenario_cfg = {} # ensure that resource_helper caches
- self.assertTrue(rfc2544_resource_helper.latency)
+ self.assertIsNone(self.rfc2544_resource_helper._latency)
+ self.assertTrue(self.rfc2544_resource_helper.latency)
+ self.assertTrue(self.rfc2544_resource_helper._latency)
+ # ensure that resource_helper caches
+ self.scenario_helper.scenario_cfg = {}
+ self.assertTrue(self.rfc2544_resource_helper.latency)
def test_property_latency_default(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_2
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_2
- self.assertFalse(rfc2544_resource_helper.latency)
+ self.assertFalse(self.rfc2544_resource_helper.latency)
def test_property_correlated_traffic(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_1
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_1
- self.assertIsNone(rfc2544_resource_helper._correlated_traffic)
- self.assertTrue(rfc2544_resource_helper.correlated_traffic)
- self.assertTrue(rfc2544_resource_helper._correlated_traffic)
- scenario_helper.scenario_cfg = {} # ensure that resource_helper caches
- self.assertTrue(rfc2544_resource_helper.correlated_traffic)
+ self.assertIsNone(self.rfc2544_resource_helper._correlated_traffic)
+ self.assertTrue(self.rfc2544_resource_helper.correlated_traffic)
+ self.assertTrue(self.rfc2544_resource_helper._correlated_traffic)
+ # ensure that resource_helper caches
+ self.scenario_helper.scenario_cfg = {}
+ self.assertTrue(self.rfc2544_resource_helper.correlated_traffic)
def test_property_correlated_traffic_default(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = self.SCENARIO_CFG_2
- rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
+ self.scenario_helper.scenario_cfg = self.SCENARIO_CFG_2
- self.assertFalse(rfc2544_resource_helper.correlated_traffic)
+ self.assertFalse(self.rfc2544_resource_helper.correlated_traffic)
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, *_):
- vnfd_helper = mock.Mock()
- ssh_helper = mock.Mock()
- ssh_helper.join_bin_path.return_value = 'joined_path'
- ssh_helper.execute.return_value = 1, 'bad output', 'error output'
- ssh_helper.put.return_value = None
- sample_vnf_deploy_helper = SampleVNFDeployHelper(vnfd_helper, ssh_helper)
-
- self.assertIsNone(sample_vnf_deploy_helper.deploy_vnfs('name1'))
- sample_vnf_deploy_helper.DISABLE_DEPLOY = True
- self.assertEqual(ssh_helper.execute.call_count, 5)
- ssh_helper.put.assert_called_once()
-
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
- @mock.patch('subprocess.check_output')
- def test_deploy_vnfs(self, *args):
- vnfd_helper = mock.Mock()
- ssh_helper = mock.Mock()
- ssh_helper.join_bin_path.return_value = 'joined_path'
- ssh_helper.execute.return_value = 1, 'bad output', 'error output'
- ssh_helper.put.return_value = None
- sample_vnf_deploy_helper = SampleVNFDeployHelper(vnfd_helper, ssh_helper)
- sample_vnf_deploy_helper.DISABLE_DEPLOY = False
-
- self.assertIsNone(sample_vnf_deploy_helper.deploy_vnfs('name1'))
- self.assertEqual(ssh_helper.execute.call_count, 5)
- ssh_helper.put.assert_called_once()
-
- @mock.patch('subprocess.check_output')
- def test_deploy_vnfs_early_success(self, *args):
- vnfd_helper = mock.Mock()
- ssh_helper = mock.Mock()
- ssh_helper.join_bin_path.return_value = 'joined_path'
- ssh_helper.execute.return_value = 0, 'output', ''
- ssh_helper.put.return_value = None
- sample_vnf_deploy_helper = SampleVNFDeployHelper(vnfd_helper, ssh_helper)
- sample_vnf_deploy_helper.DISABLE_DEPLOY = False
+ def setUp(self):
+ self._mock_time_sleep = mock.patch.object(time, 'sleep')
+ self.mock_time_sleep = self._mock_time_sleep.start()
+ self._mock_check_output = mock.patch.object(subprocess, 'check_output')
+ self.mock_check_output = self._mock_check_output.start()
+ self.addCleanup(self._stop_mocks)
- self.assertIsNone(sample_vnf_deploy_helper.deploy_vnfs('name1'))
- ssh_helper.execute.assert_called_once()
- ssh_helper.put.assert_not_called()
+ self.ssh_helper = mock.Mock()
+ self.sample_vnf_deploy_helper = sample_vnf.SampleVNFDeployHelper(
+ mock.Mock(), self.ssh_helper)
+ self.ssh_helper.join_bin_path.return_value = 'joined_path'
+ self.ssh_helper.put.return_value = None
+
+ def _stop_mocks(self):
+ self._mock_time_sleep.stop()
+ self._mock_check_output.stop()
+
+ def test_deploy_vnfs_disabled(self):
+ self.ssh_helper.execute.return_value = 1, 'bad output', 'error output'
+
+ self.sample_vnf_deploy_helper.deploy_vnfs('name1')
+ self.sample_vnf_deploy_helper.DISABLE_DEPLOY = True
+ self.assertEqual(self.ssh_helper.execute.call_count, 5)
+ self.ssh_helper.put.assert_called_once()
+
+ def test_deploy_vnfs(self):
+ self.ssh_helper.execute.return_value = 1, 'bad output', 'error output'
+ self.sample_vnf_deploy_helper.DISABLE_DEPLOY = False
+
+ self.sample_vnf_deploy_helper.deploy_vnfs('name1')
+ self.assertEqual(self.ssh_helper.execute.call_count, 5)
+ self.ssh_helper.put.assert_called_once()
+
+ def test_deploy_vnfs_early_success(self):
+ self.ssh_helper.execute.return_value = 0, 'output', ''
+ self.sample_vnf_deploy_helper.DISABLE_DEPLOY = False
+
+ self.sample_vnf_deploy_helper.deploy_vnfs('name1')
+ self.ssh_helper.execute.assert_called_once()
+ self.ssh_helper.put.assert_not_called()
class TestScenarioHelper(unittest.TestCase):
+ def setUp(self):
+ self.scenario_helper = sample_vnf.ScenarioHelper('name1')
+
def test_property_task_path(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = {
+ self.scenario_helper.scenario_cfg = {
'task_path': 'my_path',
}
- self.assertEqual(scenario_helper.task_path, 'my_path')
+ self.assertEqual(self.scenario_helper.task_path, 'my_path')
def test_property_nodes(self):
nodes = ['node1', 'node2']
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = {
+ self.scenario_helper.scenario_cfg = {
'nodes': nodes,
}
- self.assertEqual(scenario_helper.nodes, nodes)
+ self.assertEqual(self.scenario_helper.nodes, nodes)
def test_property_all_options(self):
data = {
@@ -1343,30 +1126,27 @@ class TestScenarioHelper(unittest.TestCase):
},
'name2': {}
}
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = {
+ self.scenario_helper.scenario_cfg = {
'options': data,
}
- self.assertDictEqual(scenario_helper.all_options, data)
+ self.assertDictEqual(self.scenario_helper.all_options, data)
def test_property_options(self):
data = {
'key1': 'value1',
'key2': 'value2',
}
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = {
+ self.scenario_helper.scenario_cfg = {
'options': {
'name1': data,
},
}
- self.assertDictEqual(scenario_helper.options, data)
+ self.assertDictEqual(self.scenario_helper.options, data)
def test_property_vnf_cfg(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = {
+ self.scenario_helper.scenario_cfg = {
'options': {
'name1': {
'vnf_config': 'my_config',
@@ -1374,25 +1154,24 @@ class TestScenarioHelper(unittest.TestCase):
},
}
- self.assertEqual(scenario_helper.vnf_cfg, 'my_config')
+ self.assertEqual(self.scenario_helper.vnf_cfg, 'my_config')
def test_property_vnf_cfg_default(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = {
+ self.scenario_helper.scenario_cfg = {
'options': {
'name1': {},
},
}
- self.assertDictEqual(scenario_helper.vnf_cfg, ScenarioHelper.DEFAULT_VNF_CFG)
+ self.assertEqual(self.scenario_helper.vnf_cfg,
+ sample_vnf.ScenarioHelper.DEFAULT_VNF_CFG)
def test_property_topology(self):
- scenario_helper = ScenarioHelper('name1')
- scenario_helper.scenario_cfg = {
+ self.scenario_helper.scenario_cfg = {
'topology': 'my_topology',
}
- self.assertEqual(scenario_helper.topology, 'my_topology')
+ self.assertEqual(self.scenario_helper.topology, 'my_topology')
class TestSampleVnf(unittest.TestCase):
@@ -1515,7 +1294,7 @@ class TestSampleVnf(unittest.TestCase):
}
def test___init__(self):
- sample_vnf = SampleVNF('vnf1', self.VNFD_0, 'task_id')
+ sample_vnf = SampleVNF('vnf1', self.VNFD_0)
self.assertEqual(sample_vnf.name, 'vnf1')
self.assertDictEqual(sample_vnf.vnfd_helper, self.VNFD_0)
@@ -1533,8 +1312,7 @@ class TestSampleVnf(unittest.TestCase):
class MyResourceHelper(ResourceHelper):
pass
- sample_vnf = SampleVNF('vnf1', self.VNFD_0, 'task_id',
- MySetupEnvHelper, MyResourceHelper)
+ sample_vnf = SampleVNF('vnf1', self.VNFD_0, MySetupEnvHelper, MyResourceHelper)
self.assertEqual(sample_vnf.name, 'vnf1')
self.assertDictEqual(sample_vnf.vnfd_helper, self.VNFD_0)
@@ -1548,7 +1326,7 @@ class TestSampleVnf(unittest.TestCase):
@mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.Process')
def test__start_vnf(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf1', vnfd)
sample_vnf._run = mock.Mock()
self.assertIsNone(sample_vnf.queue_wrapper)
@@ -1567,7 +1345,7 @@ class TestSampleVnf(unittest.TestCase):
}
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf1', vnfd)
sample_vnf.scenario_helper.scenario_cfg = {
'nodes': {sample_vnf.name: 'mock'}
}
@@ -1611,7 +1389,7 @@ class TestSampleVnf(unittest.TestCase):
'plugin1': {'param': 1}}}
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf__0', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf__0', vnfd)
sample_vnf._update_collectd_options(scenario_cfg, context_cfg)
self.assertEqual(sample_vnf.setup_helper.collectd_options, expected)
@@ -1638,7 +1416,7 @@ class TestSampleVnf(unittest.TestCase):
'plugin1': {'param': 1}}}
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf1', vnfd)
sample_vnf._update_options(options2, options1)
self.assertEqual(options2, expected)
@@ -1660,7 +1438,7 @@ class TestSampleVnf(unittest.TestCase):
]
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ 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)
@@ -1677,6 +1455,39 @@ class TestSampleVnf(unittest.TestCase):
self.assertEqual(sample_vnf.wait_for_instantiate(), 0)
+ @mock.patch.object(time, 'sleep')
+ @mock.patch.object(ssh, 'SSH')
+ def test_wait_for_initialize(self, ssh, *args):
+ test_base.mock_ssh(ssh, exec_result=(1, "", ""))
+ queue_get_list = [
+ 'some output',
+ 'pipeline> ',
+ 'run non_existent_script_name',
+ 'Cannot open file "non_existent_script_name"'
+ ]
+ queue_size_list = [
+ 0,
+ len(queue_get_list[0]),
+ 0,
+ len(queue_get_list[1]),
+ len(queue_get_list[2]),
+ 0,
+ len(queue_get_list[3])
+ ]
+ 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._vnf_process = mock.Mock()
+ sample_vnf._vnf_process.exitcode = 0
+ sample_vnf._vnf_process._is_alive.return_value = 1
+ sample_vnf.queue_wrapper = mock.Mock()
+ sample_vnf.q_in = mock.Mock()
+ sample_vnf.q_out = mock.Mock()
+ sample_vnf.q_out.qsize.side_effect = iter(queue_size_list)
+ sample_vnf.q_out.get.side_effect = iter(queue_get_list)
+ sample_vnf.wait_for_initialize()
+
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time")
def test_vnf_execute_with_queue_data(self, *args):
queue_size_list = [
@@ -1691,7 +1502,7 @@ class TestSampleVnf(unittest.TestCase):
]
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf1', vnfd)
sample_vnf.APP_NAME = 'sample1'
sample_vnf.q_out = mock.Mock()
sample_vnf.q_out.qsize.side_effect = iter(queue_size_list)
@@ -1701,7 +1512,7 @@ class TestSampleVnf(unittest.TestCase):
def test_terminate_without_vnf_process(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf1', vnfd)
sample_vnf.APP_NAME = 'sample1'
sample_vnf.vnf_execute = mock.Mock()
sample_vnf.ssh_helper = mock.Mock()
@@ -1712,7 +1523,7 @@ class TestSampleVnf(unittest.TestCase):
def test_get_stats(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf1', vnfd)
sample_vnf.APP_NAME = 'sample1'
sample_vnf.APP_WORD = 'sample1'
sample_vnf.vnf_execute = mock.Mock(return_value='the stats')
@@ -1722,7 +1533,7 @@ class TestSampleVnf(unittest.TestCase):
@mock.patch.object(ctx_base.Context, 'get_physical_node_from_server', return_value='mock_node')
def test_collect_kpi(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf1', vnfd)
sample_vnf.scenario_helper.scenario_cfg = {
'nodes': {sample_vnf.name: "mock"}
}
@@ -1750,7 +1561,7 @@ class TestSampleVnf(unittest.TestCase):
@mock.patch.object(ctx_base.Context, 'get_physical_node_from_server', return_value='mock_node')
def test_collect_kpi_default(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf1', vnfd)
sample_vnf.scenario_helper.scenario_cfg = {
'nodes': {sample_vnf.name: "mock"}
}
@@ -1769,7 +1580,7 @@ class TestSampleVnf(unittest.TestCase):
def test_scale(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf1', vnfd)
self.assertRaises(y_exceptions.FunctionNotImplemented,
sample_vnf.scale)
@@ -1777,7 +1588,7 @@ class TestSampleVnf(unittest.TestCase):
test_cmd = 'test cmd'
run_kwargs = {'arg1': 'val1', 'arg2': 'val2'}
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
+ sample_vnf = SampleVNF('vnf1', vnfd)
sample_vnf.ssh_helper = mock.Mock()
sample_vnf.setup_helper = mock.Mock()
with mock.patch.object(sample_vnf, '_build_config',
@@ -1913,30 +1724,30 @@ class TestSampleVNFTrafficGen(unittest.TestCase):
}
def test__check_status(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
with self.assertRaises(NotImplementedError):
sample_vnf_tg._check_status()
def test_listen_traffic(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
sample_vnf_tg.listen_traffic(mock.Mock())
def test_verify_traffic(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
sample_vnf_tg.verify_traffic(mock.Mock())
def test_terminate(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
sample_vnf_tg._traffic_process = mock.Mock()
sample_vnf_tg._tg_process = mock.Mock()
sample_vnf_tg.terminate()
def test__wait_for_process(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
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:
@@ -1947,14 +1758,14 @@ class TestSampleVNFTrafficGen(unittest.TestCase):
mock_status.assert_called_once()
def test__wait_for_process_not_alive(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
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, 'task_id')
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
with mock.patch.object(sample_vnf_tg, '_check_status',
side_effect=[1, 0]) as mock_status, \
mock.patch.object(sample_vnf_tg,
@@ -1966,6 +1777,6 @@ class TestSampleVNFTrafficGen(unittest.TestCase):
mock_status.assert_has_calls([mock.call(), mock.call()])
def test_scale(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
self.assertRaises(y_exceptions.FunctionNotImplemented,
sample_vnf_tg.scale)