# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-#
from multiprocessing import Process, Queue
-import os
import time
import mock
-from six.moves import configparser
import unittest
-from yardstick.tests import STL_MOCKS
-from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import FileAbsPath
-from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
-from yardstick.network_services.vnf_generic.vnf.base import QueueFileWrapper
-from yardstick.network_services.vnf_generic.vnf.base import VnfdHelper
-
-
-SSH_HELPER = 'yardstick.network_services.vnf_generic.vnf.sample_vnf.VnfSshHelper'
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
-
-if stl_patch:
- from yardstick.network_services.vnf_generic.vnf.vpe_vnf import ConfigCreate
- from yardstick.network_services.nfvi.resource import ResourceProfile
- from yardstick.network_services.vnf_generic.vnf.vpe_vnf import \
- VpeApproxVnf, VpeApproxSetupEnvHelper
+from yardstick.benchmark.contexts import base as ctx_base
+from yardstick.network_services.nfvi.resource import ResourceProfile
+from yardstick.network_services.vnf_generic.vnf import base as vnf_base
+from yardstick.network_services.vnf_generic.vnf import sample_vnf
+from yardstick.network_services.vnf_generic.vnf import vpe_vnf
+from yardstick.tests.unit.network_services.vnf_generic.vnf import test_base
TEST_FILE_YAML = 'nsb_test_case.yaml'
PING_OUTPUT_1 = "Pkts in: 101\r\n\tPkts dropped by AH: 100\r\n\tPkts dropped by other: 100"
-MODULE_PATH = FileAbsPath(__file__)
+MODULE_PATH = test_base.FileAbsPath(__file__)
get_file_abspath = MODULE_PATH.get_path
}
def test___init__(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- config_create = ConfigCreate(vnfd_helper, 2)
+ vnfd_helper = vnf_base.VnfdHelper(self.VNFD_0)
+ config_create = vpe_vnf.ConfigCreate(vnfd_helper, 2)
self.assertEqual(config_create.uplink_ports, ['xe0'])
self.assertEqual(config_create.downlink_ports, ['xe1'])
self.assertEqual(config_create.socket, 2)
- def test_dpdk_port_to_link_id(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- config_create = ConfigCreate(vnfd_helper, 2)
- self.assertEqual(config_create.dpdk_port_to_link_id_map, {'xe0': 0, 'xe1': 1})
-
- def test_vpe_initialize(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- config_create = ConfigCreate(vnfd_helper, 2)
- config = configparser.ConfigParser()
- config_create.vpe_initialize(config)
- self.assertEqual(config.get('EAL', 'log_level'), '0')
- self.assertEqual(config.get('PIPELINE0', 'type'), 'MASTER')
- self.assertEqual(config.get('PIPELINE0', 'core'), 's2C0')
- self.assertEqual(config.get('MEMPOOL0', 'pool_size'), '256K')
- self.assertEqual(config.get('MEMPOOL1', 'pool_size'), '2M')
-
- def test_vpe_rxq(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- config_create = ConfigCreate(vnfd_helper, 2)
- config = configparser.ConfigParser()
- config_create.downlink_ports = ['xe0']
- config_create.vpe_rxq(config)
- self.assertEqual(config.get('RXQ0.0', 'mempool'), 'MEMPOOL1')
-
- def test_get_sink_swq(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- config_create = ConfigCreate(vnfd_helper, 2)
- config = configparser.ConfigParser()
- config.add_section('PIPELINE0')
- config.set('PIPELINE0', 'key1', 'value1')
- config.set('PIPELINE0', 'key2', 'value2 SINK')
- config.set('PIPELINE0', 'key3', 'TM value3')
- config.set('PIPELINE0', 'key4', 'value4')
- config.set('PIPELINE0', 'key5', 'the SINK value5')
-
- self.assertEqual(config_create.get_sink_swq(config, 'PIPELINE0', 'key1', 5), 'SWQ-1')
- self.assertEqual(config_create.get_sink_swq(config, 'PIPELINE0', 'key2', 5), 'SWQ-1 SINK0')
- self.assertEqual(config_create.get_sink_swq(config, 'PIPELINE0', 'key3', 5), 'SWQ-1 TM5')
- config_create.sw_q += 1
- self.assertEqual(config_create.get_sink_swq(config, 'PIPELINE0', 'key4', 5), 'SWQ0')
- self.assertEqual(config_create.get_sink_swq(config, 'PIPELINE0', 'key5', 5), 'SWQ0 SINK1')
-
def test_generate_vpe_script(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- vpe_config_vnf = ConfigCreate(vnfd_helper, 2)
+ vnfd_helper = vnf_base.VnfdHelper(self.VNFD_0)
+ vpe_config_vnf = vpe_vnf.ConfigCreate(vnfd_helper, 2)
intf = [
{
"name": 'xe1',
self.assertIsInstance(result, str)
self.assertNotEqual(result, '')
- def test_create_vpe_config(self):
- vnfd_helper = VnfdHelper(self.VNFD_0)
- config_create = ConfigCreate(vnfd_helper, 23)
- config_create.downlink_ports = ['xe1']
- config_create.uplink_ports = ['xe1']
- curr_path = os.path.dirname(os.path.abspath(__file__))
- vpe_cfg = "samples/vnf_samples/nsut/vpe/vpe_config"
- vnf_cfg = os.path.join(curr_path, "../../../../..", vpe_cfg)
- config_create.create_vpe_config(vnf_cfg)
- os.system("git checkout -- %s" % vnf_cfg)
-
class TestVpeApproxVnf(unittest.TestCase):
'interfaces': {
'xe0': {
'local_iface_name': 'ens513f0',
- 'vld_id': VpeApproxVnf.DOWNLINK,
+ 'vld_id': vpe_vnf.VpeApproxVnf.DOWNLINK,
'netmask': '255.255.255.0',
'local_ip': '152.16.40.20',
'dst_mac': '00:00:00:00:00:01',
'interfaces': {
'xe0': {
'local_iface_name': 'ens785f0',
- 'vld_id': VpeApproxVnf.UPLINK,
+ 'vld_id': vpe_vnf.VpeApproxVnf.UPLINK,
'netmask': '255.255.255.0',
'local_ip': '152.16.100.20',
'dst_mac': '00:00:00:00:00:02',
'interfaces': {
'xe0': {
'local_iface_name': 'ens786f0',
- 'vld_id': VpeApproxVnf.UPLINK,
+ 'vld_id': vpe_vnf.VpeApproxVnf.UPLINK,
'netmask': '255.255.255.0',
'local_ip': '152.16.100.19',
'dst_mac': '00:00:00:00:00:04',
},
'xe1': {
'local_iface_name': 'ens786f1',
- 'vld_id': VpeApproxVnf.DOWNLINK,
+ 'vld_id': vpe_vnf.VpeApproxVnf.DOWNLINK,
'netmask': '255.255.255.0',
'local_ip': '152.16.40.19',
'dst_mac': '00:00:00:00:00:03',
self._mock_time_sleep.stop()
def test___init__(self):
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
self.assertIsNone(vpe_approx_vnf._vnf_process)
- @mock.patch(SSH_HELPER)
- def test_collect_kpi_sa_not_running(self, ssh):
- mock_ssh(ssh)
+ @mock.patch.object(ctx_base.Context, 'get_physical_node_from_server',
+ return_value='mock_node')
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
+ def test_collect_kpi_sa_not_running(self, ssh, *args):
+ test_base.mock_ssh(ssh)
resource = mock.Mock(autospec=ResourceProfile)
resource.check_if_system_agent_running.return_value = 1, ''
resource.amqp_collect_nfvi_kpi.return_value = {'foo': 234}
resource.check_if_system_agent_running.return_value = (1, None)
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
+ vpe_approx_vnf.scenario_helper.scenario_cfg = {
+ 'nodes': {vpe_approx_vnf.name: "mock"}
+ }
vpe_approx_vnf.q_in = mock.MagicMock()
vpe_approx_vnf.q_out = mock.MagicMock()
vpe_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
vpe_approx_vnf.resource_helper.resource = resource
expected = {
+ 'physical_node': 'mock_node',
'pkt_in_down_stream': 0,
'pkt_in_up_stream': 0,
'pkt_drop_down_stream': 0,
}
self.assertEqual(vpe_approx_vnf.collect_kpi(), expected)
- @mock.patch(SSH_HELPER)
- def test_collect_kpi_sa_running(self, ssh):
- mock_ssh(ssh)
+ @mock.patch.object(ctx_base.Context, 'get_physical_node_from_server',
+ return_value='mock_node')
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
+ def test_collect_kpi_sa_running(self, ssh, *args):
+ test_base.mock_ssh(ssh)
resource = mock.Mock(autospec=ResourceProfile)
resource.check_if_system_agent_running.return_value = 0, '1234'
resource.amqp_collect_nfvi_kpi.return_value = {'foo': 234}
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
+ vpe_approx_vnf.scenario_helper.scenario_cfg = {
+ 'nodes': {vpe_approx_vnf.name: "mock"}
+ }
vpe_approx_vnf.q_in = mock.MagicMock()
vpe_approx_vnf.q_out = mock.MagicMock()
vpe_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
vpe_approx_vnf.resource_helper.resource = resource
expected = {
+ 'physical_node': 'mock_node',
'pkt_in_down_stream': 0,
'pkt_in_up_stream': 0,
'pkt_drop_down_stream': 0,
}
self.assertEqual(vpe_approx_vnf.collect_kpi(), expected)
- @mock.patch(SSH_HELPER)
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
def test_vnf_execute(self, ssh):
- mock_ssh(ssh)
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ test_base.mock_ssh(ssh)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
vpe_approx_vnf.q_in = mock.MagicMock()
vpe_approx_vnf.q_out = mock.MagicMock()
vpe_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
self.assertEqual(vpe_approx_vnf.vnf_execute("quit", 0), '')
- @mock.patch(SSH_HELPER)
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
def test_run_vpe(self, ssh):
- mock_ssh(ssh)
+ test_base.mock_ssh(ssh)
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
vpe_approx_vnf.tc_file_name = get_file_abspath(TEST_FILE_YAML)
vpe_approx_vnf.vnf_cfg = {
'lb_config': 'SW',
self.assertIsNone(vpe_approx_vnf._run())
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.MultiPortConfig")
- @mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.Context")
@mock.patch("yardstick.network_services.vnf_generic.vnf.vpe_vnf.ConfigCreate")
@mock.patch("six.moves.builtins.open")
- @mock.patch(SSH_HELPER)
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
def test_build_config(self, ssh, *args):
- mock_ssh(ssh)
- vpe_approx_vnf = VpeApproxSetupEnvHelper(mock.MagicMock(),
- mock.MagicMock(), mock.MagicMock())
+ test_base.mock_ssh(ssh)
+ vpe_approx_vnf = vpe_vnf.VpeApproxSetupEnvHelper(
+ mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
vpe_approx_vnf.tc_file_name = get_file_abspath(TEST_FILE_YAML)
vpe_approx_vnf.generate_port_pairs = mock.Mock()
vpe_approx_vnf.vnf_cfg = {
expected = 'sudo tool_path -p 0x3 -f /tmp/vpe_config -s /tmp/vpe_script --hwlb 3'
self.assertEqual(vpe_approx_vnf.build_config(), expected)
- @mock.patch(SSH_HELPER)
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
def test_wait_for_instantiate(self, ssh):
- mock_ssh(ssh)
+ test_base.mock_ssh(ssh)
mock_process = mock.Mock(autospec=Process)
mock_process.is_alive.return_value = True
mock_resource = mock.MagicMock()
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
vpe_approx_vnf._vnf_process = mock_process
vpe_approx_vnf.q_out = mock_q_out
- vpe_approx_vnf.queue_wrapper = mock.Mock(autospec=QueueFileWrapper)
+ vpe_approx_vnf.queue_wrapper = mock.Mock(
+ autospec=vnf_base.QueueFileWrapper)
vpe_approx_vnf.resource_helper.resource = mock_resource
vpe_approx_vnf.q_out.put("pipeline>")
self.assertEqual(vpe_approx_vnf.wait_for_instantiate(), 432)
- @mock.patch(SSH_HELPER)
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
def test_wait_for_instantiate_fragmented(self, ssh):
- mock_ssh(ssh)
+ test_base.mock_ssh(ssh)
mock_process = mock.Mock(autospec=Process)
mock_process.is_alive.return_value = True
mock_resource = mock.MagicMock()
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
vpe_approx_vnf._vnf_process = mock_process
vpe_approx_vnf.q_out = mock_q_out
- vpe_approx_vnf.queue_wrapper = mock.Mock(autospec=QueueFileWrapper)
+ vpe_approx_vnf.queue_wrapper = mock.Mock(
+ autospec=vnf_base.QueueFileWrapper)
vpe_approx_vnf.resource_helper.resource = mock_resource
self.assertEqual(vpe_approx_vnf.wait_for_instantiate(), 432)
- @mock.patch(SSH_HELPER)
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
def test_wait_for_instantiate_crash(self, ssh):
- mock_ssh(ssh, exec_result=(1, "", ""))
+ test_base.mock_ssh(ssh, exec_result=(1, "", ""))
mock_process = mock.Mock(autospec=Process)
mock_process.is_alive.return_value = False
mock_resource = mock.MagicMock()
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
vpe_approx_vnf._vnf_process = mock_process
vpe_approx_vnf.resource_helper.resource = mock_resource
self.assertIn('VNF process died', str(raised.exception))
- @mock.patch(SSH_HELPER)
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
def test_wait_for_instantiate_panic(self, ssh):
- mock_ssh(ssh, exec_result=(1, "", ""))
+ test_base.mock_ssh(ssh, exec_result=(1, "", ""))
mock_process = mock.Mock(autospec=Process)
mock_process.is_alive.return_value = True
mock_resource = mock.MagicMock()
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
vpe_approx_vnf._vnf_process = mock_process
vpe_approx_vnf.resource_helper.resource = mock_resource
self.assertIn('Error starting', str(raised.exception))
- @mock.patch(SSH_HELPER)
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
def test_wait_for_instantiate_panic_fragmented(self, ssh):
- mock_ssh(ssh, exec_result=(1, "", ""))
+ test_base.mock_ssh(ssh, exec_result=(1, "", ""))
mock_process = mock.Mock(autospec=Process)
mock_process.is_alive.return_value = True
mock_resource = mock.MagicMock()
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
vpe_approx_vnf._vnf_process = mock_process
vpe_approx_vnf.q_out = mock_q_out
vpe_approx_vnf.resource_helper.resource = mock_resource
self.assertIn('Error starting', str(raised.exception))
- @mock.patch(SSH_HELPER)
+ @mock.patch.object(sample_vnf, 'VnfSshHelper')
def test_terminate(self, ssh):
- mock_ssh(ssh)
+ test_base.mock_ssh(ssh)
- vpe_approx_vnf = VpeApproxVnf(NAME, self.VNFD_0)
+ vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0, 'task_id')
vpe_approx_vnf._vnf_process = mock.MagicMock()
vpe_approx_vnf._resource_collect_stop = mock.Mock()
vpe_approx_vnf.resource_helper = mock.MagicMock()
Code Review / yardstick.git / blobdiff