X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Funit%2Fnetwork_services%2Fvnf_generic%2Fvnf%2Ftest_udp_replay.py;h=cda3852fe04162fa058a81aac5f9879a0b5a2cf5;hb=d2f89835fa08f88f0100dae2996c49302ee25614;hp=c4ced30fec699900608706909cf7263481e5760f;hpb=a65eb4e43179dc4d75d0e2cf158cc73b9ad9a3b8;p=yardstick.git diff --git a/tests/unit/network_services/vnf_generic/vnf/test_udp_replay.py b/tests/unit/network_services/vnf_generic/vnf/test_udp_replay.py index c4ced30fe..cda3852fe 100644 --- a/tests/unit/network_services/vnf_generic/vnf/test_udp_replay.py +++ b/tests/unit/network_services/vnf_generic/vnf/test_udp_replay.py @@ -15,16 +15,16 @@ # limitations under the License. # -from __future__ import absolute_import - import unittest import mock import os from tests.unit import STL_MOCKS -SSH_HELPER = 'yardstick.network_services.vnf_generic.vnf.sample_vnf.VnfSshHelper' +from tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh +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() @@ -33,7 +33,6 @@ if stl_patch: from yardstick.network_services.vnf_generic.vnf.udp_replay import UdpReplayApproxVnf from yardstick.network_services.vnf_generic.vnf.sample_vnf import ScenarioHelper -from tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh TEST_FILE_YAML = 'nsb_test_case.yaml' @@ -73,10 +72,12 @@ class TestUdpReplayApproxVnf(unittest.TestCase): 'local_ip': '152.16.100.19', 'local_mac': '00:00:00:00:00:02', 'vpci': '0000:05:00.0', - 'dpdk_port_num': '0', + 'dpdk_port_num': 0, 'netmask': '255.255.255.0', 'dst_ip': '152.16.100.20', 'type': 'PCI-PASSTHROUGH', + 'vld_id': 'uplink_0', + 'ifname': 'xe0', }, 'vnfd-connection-point-ref': 'xe0', 'name': 'xe0', @@ -90,10 +91,12 @@ class TestUdpReplayApproxVnf(unittest.TestCase): 'local_ip': '152.16.40.19', 'local_mac': '00:00:00:00:00:01', 'vpci': '0000:05:00.1', - 'dpdk_port_num': '1', + 'dpdk_port_num': 1, 'netmask': '255.255.255.0', 'dst_ip': '152.16.40.20', 'type': 'PCI-PASSTHROUGH', + 'vld_id': 'downlink_0', + 'ifname': 'xe1', }, 'vnfd-connection-point-ref': 'xe1', 'name': 'xe1', @@ -195,7 +198,7 @@ class TestUdpReplayApproxVnf(unittest.TestCase): "interfaces": { "xe0": { "local_iface_name": "ens786f0", - "vld_id": "private", + "vld_id": UdpReplayApproxVnf.UPLINK, "netmask": "255.255.255.0", "vpci": "0000:05:00.0", "local_ip": "152.16.100.19", @@ -207,7 +210,7 @@ class TestUdpReplayApproxVnf(unittest.TestCase): }, "xe1": { "local_iface_name": "ens786f1", - "vld_id": "public", + "vld_id": UdpReplayApproxVnf.DOWNLINK, "netmask": "255.255.255.0", "vpci": "0000:05:00.1", "local_ip": "152.16.40.19", @@ -263,7 +266,7 @@ class TestUdpReplayApproxVnf(unittest.TestCase): "interfaces": { "xe0": { "local_iface_name": "ens513f0", - "vld_id": "public", + "vld_id": UdpReplayApproxVnf.DOWNLINK, "netmask": "255.255.255.0", "vpci": "0000:02:00.0", "local_ip": "152.16.40.20", @@ -297,7 +300,7 @@ class TestUdpReplayApproxVnf(unittest.TestCase): "interfaces": { "xe0": { "local_iface_name": "ens785f0", - "vld_id": "private", + "vld_id": UdpReplayApproxVnf.UPLINK, "netmask": "255.255.255.0", "vpci": "0000:05:00.0", "local_ip": "152.16.100.20", @@ -324,13 +327,13 @@ class TestUdpReplayApproxVnf(unittest.TestCase): } } - def test___init__(self, _): + def test___init__(self, *args): udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0) self.assertIsNone(udp_replay_approx_vnf._vnf_process) @mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time") @mock.patch(SSH_HELPER) - def test_collect_kpi(self, ssh, mock_time, _): + def test_collect_kpi(self, ssh, *args): mock_ssh(ssh) vnfd = self.VNFD_0 @@ -342,23 +345,14 @@ class TestUdpReplayApproxVnf(unittest.TestCase): udp_replay_approx_vnf.q_in = mock.MagicMock() udp_replay_approx_vnf.q_out = mock.MagicMock() udp_replay_approx_vnf.q_out.qsize = mock.Mock(return_value=0) - udp_replay_approx_vnf.all_ports = [0, 1] + udp_replay_approx_vnf.all_ports = ["xe0", "xe1"] udp_replay_approx_vnf.get_stats = mock.Mock(return_value=result) result = {'collect_stats': {}, 'packets_dropped': 0, 'packets_fwd': 14748451, 'packets_in': 14748472} self.assertEqual(result, udp_replay_approx_vnf.collect_kpi()) - @mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time") - @mock.patch(SSH_HELPER) - def test_vnf_execute_command(self, ssh, mock_time, _): - mock_ssh(ssh) - - udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0) - cmd = "quit" - self.assertEqual("", udp_replay_approx_vnf.vnf_execute(cmd)) - @mock.patch(SSH_HELPER) - def test_get_stats(self, ssh, _): + def test_get_stats(self, ssh, *args): mock_ssh(ssh) udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0) @@ -378,10 +372,9 @@ class TestUdpReplayApproxVnf(unittest.TestCase): file_path = os.path.join(curr_path, filename) return file_path - @mock.patch('yardstick.network_services.vnf_generic.vnf.udp_replay.open') @mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.Context") @mock.patch(SSH_HELPER) - def test__build_config(self, ssh, mock_context, *_): + def test__build_config(self, ssh, mock_context, *args): mock_ssh(ssh) udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0) @@ -389,29 +382,26 @@ class TestUdpReplayApproxVnf(unittest.TestCase): udp_replay_approx_vnf.nfvi_context = mock_context udp_replay_approx_vnf.nfvi_context.attrs = {'nfvi_type': 'baremetal'} udp_replay_approx_vnf.setup_helper.bound_pci = [] - udp_replay_approx_vnf.all_ports = [0, 1] udp_replay_approx_vnf.ssh_helper.provision_tool = mock.MagicMock(return_value="tool_path") udp_replay_approx_vnf.scenario_helper = ScenarioHelper(name='vnf__1') udp_replay_approx_vnf.scenario_helper.scenario_cfg = self.SCENARIO_CFG cmd_line = udp_replay_approx_vnf._build_config() - expected = "sudo tool_path -c 0x7 -n 4 -w -- -p 0x3 --config='(0, 0, 1)(1, 0, 2)'" + expected = \ + "sudo tool_path --log-level=5 -c 0x7 -n 4 -w -- -p 0x3 --config='(0,0,1),(1,0,2)'" self.assertEqual(cmd_line, expected) @mock.patch('yardstick.network_services.vnf_generic.vnf.udp_replay.open') @mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.Context") @mock.patch(SSH_HELPER) - def test__build_pipeline_kwargs(self, ssh, mock_context, *_): + def test__build_pipeline_kwargs(self, ssh, mock_context, *args): mock_ssh(ssh) - udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0) - udp_replay_approx_vnf._build_config = mock.MagicMock() - udp_replay_approx_vnf.queue_wrapper = mock.MagicMock() udp_replay_approx_vnf.nfvi_context = mock_context udp_replay_approx_vnf.nfvi_context.attrs = {'nfvi_type': 'baremetal'} - udp_replay_approx_vnf.setup_helper.bound_pci = [] - udp_replay_approx_vnf.all_ports = [0, 1] + udp_replay_approx_vnf.setup_helper.bound_pci = ['0000:00:0.1', '0000:00:0.3'] + udp_replay_approx_vnf.all_ports = ["xe0", "xe1"] udp_replay_approx_vnf.ssh_helper.provision_tool = mock.MagicMock(return_value="tool_path") udp_replay_approx_vnf.scenario_helper = ScenarioHelper(name='vnf__1') udp_replay_approx_vnf.scenario_helper.scenario_cfg = self.SCENARIO_CFG @@ -419,21 +409,22 @@ class TestUdpReplayApproxVnf(unittest.TestCase): udp_replay_approx_vnf._build_pipeline_kwargs() self.assertEqual(udp_replay_approx_vnf.pipeline_kwargs, { - 'config': '(0, 0, 1)(1, 0, 2)', + 'config': '(0,0,1),(1,0,2)', 'cpu_mask_hex': '0x7', 'hw_csum': '', - 'ports_len_hex': '0x3', + 'port_mask_hex': '0x3', 'tool_path': 'tool_path', - 'whitelist': '' + 'whitelist': '0000:00:0.1 -w 0000:00:0.3' }) @mock.patch(SSH_HELPER) - def test_run_udp_replay(self, ssh, _): + def test_run_udp_replay(self, ssh, *args): mock_ssh(ssh) udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0) udp_replay_approx_vnf._build_config = mock.MagicMock() udp_replay_approx_vnf.queue_wrapper = mock.MagicMock() + udp_replay_approx_vnf.scenario_helper = mock.MagicMock() udp_replay_approx_vnf._run() @@ -441,7 +432,7 @@ class TestUdpReplayApproxVnf(unittest.TestCase): @mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.Context") @mock.patch(SSH_HELPER) - def test_instantiate(self, ssh, *_): + def test_instantiate(self, ssh, *args): mock_ssh(ssh) udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0) @@ -449,45 +440,37 @@ class TestUdpReplayApproxVnf(unittest.TestCase): udp_replay_approx_vnf.WAIT_TIME = 0 udp_replay_approx_vnf.setup_helper.setup_vnf_environment = mock.Mock() - self.assertIsNone(udp_replay_approx_vnf.instantiate(self.SCENARIO_CFG, - self.CONTEXT_CFG)) + udp_replay_approx_vnf.deploy_helper = mock.MagicMock() + udp_replay_approx_vnf.deploy_vnfs = mock.MagicMock() + self.assertIsNone(udp_replay_approx_vnf.instantiate(self.SCENARIO_CFG, self.CONTEXT_CFG)) udp_replay_approx_vnf._vnf_process.is_alive = mock.Mock(return_value=1) udp_replay_approx_vnf._vnf_process.exitcode = 0 - self.assertEquals(udp_replay_approx_vnf.wait_for_instantiate(), 0) + self.assertEqual(udp_replay_approx_vnf.wait_for_instantiate(), 0) @mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.Context") @mock.patch('yardstick.ssh.SSH') @mock.patch(SSH_HELPER) - def test_instantiate_panic(self, ssh, resource_ssh, *_): + def test_instantiate_panic(self, *args): udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0) udp_replay_approx_vnf.WAIT_TIME = 0 udp_replay_approx_vnf.q_out.put("some text PANIC some text") udp_replay_approx_vnf.setup_helper.setup_vnf_environment = mock.Mock() + udp_replay_approx_vnf.deploy_helper = mock.MagicMock() self.assertIsNone(udp_replay_approx_vnf.instantiate(self.SCENARIO_CFG, self.CONTEXT_CFG)) with self.assertRaises(RuntimeError): udp_replay_approx_vnf.wait_for_instantiate() - def test_scale(self, _): - udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0) - flavor = "" - - self.assertRaises(NotImplementedError, udp_replay_approx_vnf.scale, flavor) - @mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time") @mock.patch(SSH_HELPER) - def test_terminate(self, ssh, mock_time, _): + def test_terminate(self, ssh, *args): mock_ssh(ssh) udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0) udp_replay_approx_vnf._vnf_process = mock.MagicMock() udp_replay_approx_vnf._vnf_process.terminate = mock.Mock() - udp_replay_approx_vnf.used_drivers = {"01:01.0": "i40e", - "01:01.1": "i40e"} + udp_replay_approx_vnf.used_drivers = {"01:01.0": "i40e", "01:01.1": "i40e"} udp_replay_approx_vnf.dpdk_nic_bind = "dpdk_nic_bind.py" self.assertEqual(None, udp_replay_approx_vnf.terminate()) - -if __name__ == '__main__': - unittest.main()