# limitations under the License.
#
-from __future__ import absolute_import
-
import unittest
import mock
import os
from tests.unit import STL_MOCKS
+from tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
+
+from yardstick.common import utils
STLClient = mock.MagicMock()
from yardstick.network_services.vnf_generic.vnf.vfw_vnf import FWApproxVnf
from yardstick.network_services.nfvi.resource import ResourceProfile
-TEST_FILE_YAML = 'nsb_test_case.yaml'
+TEST_FILE_YAML = 'nsb_test_case.yaml'
+SSH_HELPER = 'yardstick.network_services.vnf_generic.vnf.sample_vnf.VnfSshHelper'
name = 'vnf__1'
'local_ip': '152.16.100.19',
'type': 'PCI-PASSTHROUGH',
'netmask': '255.255.255.0',
- 'dpdk_port_num': '0',
+ 'dpdk_port_num': 0,
'bandwidth': '10 Gbps',
'driver': "i40e",
'dst_ip': '152.16.100.20',
'type': 'PCI-PASSTHROUGH',
'driver': "i40e",
'netmask': '255.255.255.0',
- 'dpdk_port_num': '1',
+ 'dpdk_port_num': 1,
'bandwidth': '10 Gbps',
'dst_ip': '152.16.40.20',
'local_iface_name': 'xe1',
'ip': '1.2.1.1',
'interfaces':
{'xe0': {'local_iface_name': 'ens513f0',
- 'vld_id': 'public',
+ 'vld_id': FWApproxVnf.DOWNLINK,
'netmask': '255.255.255.0',
'local_ip': '152.16.40.20',
'dst_mac': '00:00:00:00:00:01',
'ip': '1.2.1.1',
'interfaces':
{'xe0': {'local_iface_name': 'ens785f0',
- 'vld_id': 'private',
+ 'vld_id': FWApproxVnf.UPLINK,
'netmask': '255.255.255.0',
'local_ip': '152.16.100.20',
'dst_mac': '00:00:00:00:00:02',
'ip': '1.2.1.1',
'interfaces':
{'xe0': {'local_iface_name': 'ens786f0',
- 'vld_id': 'private',
+ 'vld_id': FWApproxVnf.UPLINK,
'netmask': '255.255.255.0',
'local_ip': '152.16.100.19',
'dst_mac': '00:00:00:00:00:04',
'vpci': '0000:05:00.0',
'dpdk_port_num': 0},
'xe1': {'local_iface_name': 'ens786f1',
- 'vld_id': 'public',
+ 'vld_id': FWApproxVnf.DOWNLINK,
'netmask': '255.255.255.0',
'local_ip': '152.16.40.19',
'dst_mac': '00:00:00:00:00:03',
'password': 'r00t',
'VNF model': 'vfw_vnf.yaml'}}}
- def test___init__(self, mock_process):
+ def test___init__(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
vfw_approx_vnf = FWApproxVnf(name, vnfd)
self.assertIsNone(vfw_approx_vnf._vnf_process)
""" # noqa
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time")
- def test_collect_kpi(self, mock_time, mock_process):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh.from_node.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
- vfw_approx_vnf.q_in = mock.MagicMock()
- vfw_approx_vnf.q_out = mock.MagicMock()
- vfw_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
- vfw_approx_vnf.resource = mock.Mock(autospec=ResourceProfile)
- vfw_approx_vnf.resource_helper = mock.MagicMock(
- **{'collect_kpi.return_value': {"core": {}}})
- vfw_approx_vnf.vnf_execute = mock.Mock(return_value=self.STATS)
- result = {
- 'packets_dropped': 0,
- 'packets_fwd': 6007180,
- 'packets_in': 6007180,
- 'collect_stats': {'core': {}},
- }
- self.assertEqual(result, vfw_approx_vnf.collect_kpi())
+ @mock.patch(SSH_HELPER)
+ def test_collect_kpi(self, ssh, *args):
+ mock_ssh(ssh)
+
+ vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
+ vfw_approx_vnf = FWApproxVnf(name, vnfd)
+ vfw_approx_vnf.q_in = mock.MagicMock()
+ vfw_approx_vnf.q_out = mock.MagicMock()
+ vfw_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
+ vfw_approx_vnf.resource = mock.Mock(autospec=ResourceProfile)
+ vfw_approx_vnf.resource_helper = mock.MagicMock(
+ **{'collect_kpi.return_value': {"core": {}}})
+ vfw_approx_vnf.vnf_execute = mock.Mock(return_value=self.STATS)
+ result = {
+ 'packets_dropped': 0,
+ 'packets_fwd': 6007180,
+ 'packets_in': 6007180,
+ 'collect_stats': {'core': {}},
+ }
+ self.assertEqual(result, vfw_approx_vnf.collect_kpi())
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time")
- def test_vnf_execute_command(self, mock_time, mock_process):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh.from_node.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
- vfw_approx_vnf.q_in = mock.MagicMock()
- vfw_approx_vnf.q_out = mock.MagicMock()
- vfw_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
- cmd = "quit"
- self.assertEqual("", vfw_approx_vnf.vnf_execute(cmd))
+ @mock.patch(SSH_HELPER)
+ def test_vnf_execute_command(self, ssh, *args):
+ mock_ssh(ssh)
+
+ vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
+ vfw_approx_vnf = FWApproxVnf(name, vnfd)
+ vfw_approx_vnf.q_in = mock.MagicMock()
+ vfw_approx_vnf.q_out = mock.MagicMock()
+ vfw_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
+ cmd = "quit"
+ self.assertEqual(vfw_approx_vnf.vnf_execute(cmd), "")
- def test_get_stats(self, mock_process):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh.from_node.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
- vfw_approx_vnf.q_in = mock.MagicMock()
- vfw_approx_vnf.q_out = mock.MagicMock()
- vfw_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
- vfw_approx_vnf.vnf_execute = mock.Mock(return_value=self.STATS)
- self.assertEqual(self.STATS, vfw_approx_vnf.get_stats())
+ @mock.patch(SSH_HELPER)
+ def test_get_stats(self, ssh, *args):
+ mock_ssh(ssh)
+
+ vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
+ vfw_approx_vnf = FWApproxVnf(name, vnfd)
+ vfw_approx_vnf.q_in = mock.MagicMock()
+ vfw_approx_vnf.q_out = mock.MagicMock()
+ vfw_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
+ vfw_approx_vnf.vnf_execute = mock.Mock(return_value=self.STATS)
+ self.assertEqual(self.STATS, vfw_approx_vnf.get_stats())
def _get_file_abspath(self, filename):
curr_path = os.path.dirname(os.path.abspath(__file__))
@mock.patch("yardstick.network_services.vnf_generic.vnf.vfw_vnf.hex")
@mock.patch("yardstick.network_services.vnf_generic.vnf.vfw_vnf.eval")
@mock.patch("yardstick.network_services.vnf_generic.vnf.vfw_vnf.open")
- def test_run_vfw(self, mock_open, eval, hex, mock_process):
- 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.from_node.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
- vfw_approx_vnf._build_config = mock.MagicMock()
- vfw_approx_vnf.queue_wrapper = mock.MagicMock()
- vfw_approx_vnf.ssh_helper = mock.MagicMock()
- vfw_approx_vnf.ssh_helper.run = mock.MagicMock()
- vfw_approx_vnf.scenario_helper.scenario_cfg = self.scenario_cfg
- vfw_approx_vnf.vnf_cfg = {'lb_config': 'SW',
- 'lb_count': 1,
- 'worker_config': '1C/1T',
- 'worker_threads': 1}
- vfw_approx_vnf.all_options = {'traffic_type': '4',
- 'topology': 'nsb_test_case.yaml'}
- vfw_approx_vnf._run()
- vfw_approx_vnf.ssh_helper.run.assert_called_once()
+ @mock.patch(SSH_HELPER)
+ def test_run_vfw(self, ssh, *args):
+ mock_ssh(ssh)
+ vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
+ vfw_approx_vnf = FWApproxVnf(name, vnfd)
+ vfw_approx_vnf._build_config = mock.MagicMock()
+ vfw_approx_vnf.queue_wrapper = mock.MagicMock()
+ vfw_approx_vnf.ssh_helper = mock.MagicMock()
+ vfw_approx_vnf.ssh_helper.run = mock.MagicMock()
+ vfw_approx_vnf.scenario_helper.scenario_cfg = self.scenario_cfg
+ vfw_approx_vnf.vnf_cfg = {'lb_config': 'SW',
+ 'lb_count': 1,
+ 'worker_config': '1C/1T',
+ 'worker_threads': 1}
+ vfw_approx_vnf.all_options = {'traffic_type': '4',
+ 'topology': 'nsb_test_case.yaml'}
+ vfw_approx_vnf._run()
+ vfw_approx_vnf.ssh_helper.run.assert_called_once()
+
+ @mock.patch.object(utils, 'find_relative_file')
@mock.patch("yardstick.network_services.vnf_generic.vnf.vfw_vnf.YangModel")
- @mock.patch("yardstick.network_services.vnf_generic.vnf.vfw_vnf.find_relative_file")
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.Context")
- def test_instantiate(self, Context, mock_yang, mock_find, mock_process):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = mock.Mock(return_value=(0, "", ""))
- ssh.from_node.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
- vfw_approx_vnf.ssh_helper = ssh
- vfw_approx_vnf.deploy_helper = mock.MagicMock()
- vfw_approx_vnf.resource_helper = mock.MagicMock()
- vfw_approx_vnf._build_config = mock.MagicMock()
- self.scenario_cfg['vnf_options'] = {'acl': {'cfg': "",
- 'rules': ""}}
- self.scenario_cfg.update({"nodes": {"vnf__1": ""}})
- self.assertIsNone(vfw_approx_vnf.instantiate(self.scenario_cfg,
- self.context_cfg))
+ @mock.patch(SSH_HELPER)
+ def test_instantiate(self, ssh, *args):
+ mock_ssh(ssh)
- def test_scale(self, mock_process):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
vfw_approx_vnf = FWApproxVnf(name, vnfd)
- flavor = ""
- self.assertRaises(NotImplementedError, vfw_approx_vnf.scale, flavor)
-
- @mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time")
- def test_terminate(self, mock_time, mock_process):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh.from_node.return_value = ssh_mock
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
- vfw_approx_vnf._vnf_process = mock.MagicMock()
- vfw_approx_vnf._vnf_process.terminate = mock.Mock()
- vfw_approx_vnf.used_drivers = {"01:01.0": "i40e",
- "01:01.1": "i40e"}
- vfw_approx_vnf.vnf_execute = mock.Mock()
- vfw_approx_vnf.ssh_helper = ssh_mock
- vfw_approx_vnf.dpdk_nic_bind = "dpdk_nic_bind.py"
- vfw_approx_vnf._resource_collect_stop = mock.Mock()
- self.assertEqual(None, vfw_approx_vnf.terminate())
-
-if __name__ == '__main__':
- unittest.main()
+ vfw_approx_vnf.ssh_helper = ssh
+ vfw_approx_vnf.deploy_helper = mock.MagicMock()
+ vfw_approx_vnf.resource_helper = mock.MagicMock()
+ vfw_approx_vnf._build_config = mock.MagicMock()
+ self.scenario_cfg['vnf_options'] = {'acl': {'cfg': "",
+ 'rules': ""}}
+ self.scenario_cfg.update({"nodes": {"vnf__1": ""}})
+ self.assertIsNone(vfw_approx_vnf.instantiate(self.scenario_cfg, self.context_cfg))
Code Review / yardstick.git / blobdiff