# 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 copy import deepcopy
import mock
import six
-from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
-from yardstick.tests import STL_MOCKS
-from yardstick.benchmark.contexts.base import Context
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
-
-
-class MockError(BaseException):
+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.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
+
+
+class MockError(Exception):
pass
-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 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.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
-
-
class TestVnfSshHelper(unittest.TestCase):
VNFD_0 = {
}
@mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.LOG')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.STLError',
- new_callable=lambda: MockError)
- def test_get_stats_not_connected(self, mock_state_error, *args):
+ @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)
ssh_helper = mock.Mock()
scenario_helper = mock.Mock()
- dpdk_setup_helper = DpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
+ dpdk_setup_helper = DpdkVnfSetupEnvHelper(
+ vnfd_helper, ssh_helper, scenario_helper)
client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
- client_resource_helper.client = mock.MagicMock()
- client_resource_helper.client.get_stats.side_effect = mock_state_error
+ client_resource_helper.client = mock.Mock()
+ client_resource_helper.client.get_stats.side_effect = mock_stl_error
self.assertEqual(client_resource_helper.get_stats(), {})
client_resource_helper.client.get_stats.assert_called_once()
- def test_generate_samples(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.MagicMock()
- client_resource_helper.client.get_stats.return_value = {
- 0: {
- 'rx_pps': 5.5,
- 'tx_pps': 4.9,
- 'rx_bps': 234.78,
- 'tx_bps': 243.11,
- 'ipackets': 34251,
- 'opackets': 52342,
- },
- 1: {
- 'tx_pps': 5.9,
- 'rx_bps': 434.78,
- 'opackets': 48791,
- },
- }
-
- expected = {
- 'xe0': {
- "rx_throughput_fps": 5.5,
- "tx_throughput_fps": 4.9,
- "rx_throughput_mbps": 234.78,
- "tx_throughput_mbps": 243.11,
- "in_packets": 34251,
- "out_packets": 52342,
- },
- 'xe1': {
- "rx_throughput_fps": 0.0,
- "tx_throughput_fps": 5.9,
- "rx_throughput_mbps": 434.78,
- "tx_throughput_mbps": 0.0,
- "in_packets": 0,
- "out_packets": 48791,
- },
- }
- ports = vnfd_helper.port_nums(vnfd_helper.port_pairs.all_ports)
- result = client_resource_helper.generate_samples(ports)
- self.assertDictEqual(result, expected)
-
- def test_generate_samples_with_key(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.MagicMock()
- client_resource_helper.client.get_stats.return_value = {
- 'key_name': 'key_value',
- 0: {
- 'rx_pps': 5.5,
- 'tx_pps': 4.9,
- 'rx_bps': 234.78,
- 'tx_bps': 243.11,
- 'ipackets': 34251,
- 'opackets': 52342,
- },
- 1: {
- 'tx_pps': 5.9,
- 'rx_bps': 434.78,
- 'opackets': 48791,
- },
- }
-
- expected = {
- 'xe0': {
- 'key_name': 'key_value',
- "rx_throughput_fps": 5.5,
- "tx_throughput_fps": 4.9,
- "rx_throughput_mbps": 234.78,
- "tx_throughput_mbps": 243.11,
- "in_packets": 34251,
- "out_packets": 52342,
- },
- 'xe1': {
- 'key_name': 'key_value',
- "rx_throughput_fps": 0.0,
- "tx_throughput_fps": 5.9,
- "rx_throughput_mbps": 434.78,
- "tx_throughput_mbps": 0.0,
- "in_packets": 0,
- "out_packets": 48791,
- },
- }
- ports = vnfd_helper.port_nums(vnfd_helper.port_pairs.all_ports)
- result = client_resource_helper.generate_samples(ports, 'key_name')
- self.assertDictEqual(result, expected)
-
- def test_generate_samples_with_key_and_default(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.MagicMock()
- client_resource_helper.client.get_stats.return_value = {
- 0: {
- 'rx_pps': 5.5,
- 'tx_pps': 4.9,
- 'rx_bps': 234.78,
- 'tx_bps': 243.11,
- 'ipackets': 34251,
- 'opackets': 52342,
- },
- 1: {
- 'tx_pps': 5.9,
- 'rx_bps': 434.78,
- 'opackets': 48791,
- },
- }
-
- expected = {
- 'xe0': {
- 'key_name': 'default',
- "rx_throughput_fps": 5.5,
- "tx_throughput_fps": 4.9,
- "rx_throughput_mbps": 234.78,
- "tx_throughput_mbps": 243.11,
- "in_packets": 34251,
- "out_packets": 52342,
- },
- 'xe1': {
- 'key_name': 'default',
- "rx_throughput_fps": 0.0,
- "tx_throughput_fps": 5.9,
- "rx_throughput_mbps": 434.78,
- "tx_throughput_mbps": 0.0,
- "in_packets": 0,
- "out_packets": 48791,
- },
- }
- ports = vnfd_helper.port_nums(vnfd_helper.port_pairs.all_ports)
- result = client_resource_helper.generate_samples(ports, 'key_name', 'default')
- self.assertDictEqual(result, expected)
-
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)
+ dpdk_setup_helper = DpdkVnfSetupEnvHelper(
+ vnfd_helper, ssh_helper, scenario_helper)
client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
client_resource_helper.client = mock.Mock()
self.assertIsNone(client_resource_helper.clear_stats())
- client_resource_helper.client.clear_stats.assert_called_once()
+ self.assertEqual(
+ 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)
+ dpdk_setup_helper = DpdkVnfSetupEnvHelper(
+ vnfd_helper, ssh_helper, scenario_helper)
client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
client_resource_helper.client = mock.Mock()
self.assertIsNone(client_resource_helper.clear_stats([3, 4]))
- client_resource_helper.client.clear_stats.assert_called_once()
+ self.assertEqual(
+ client_resource_helper.client.clear_stats.call_count, 1)
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)
+ dpdk_setup_helper = DpdkVnfSetupEnvHelper(
+ vnfd_helper, ssh_helper, scenario_helper)
client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
client_resource_helper.client = mock.Mock()
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 = DpdkVnfSetupEnvHelper(
+ vnfd_helper, ssh_helper, scenario_helper)
client_resource_helper = ClientResourceHelper(dpdk_setup_helper)
client_resource_helper.client = mock.Mock()
self.assertDictEqual(result, expected)
@mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.time')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.LOG')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.STLError',
- new_callable=lambda: MockError)
- def test__connect_with_failures(self, mock_error, *args):
+ @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_error
+ client.connect.side_effect = mock_stl_error(msg='msg')
self.assertIs(client_resource_helper._connect(client), client)
self.assertIsNotNone(sample_vnf.queue_wrapper)
self.assertIsNotNone(sample_vnf._vnf_process)
+ @mock.patch.object(ctx_base.Context, 'get_context_from_server', return_value='fake_context')
@mock.patch("yardstick.ssh.SSH")
- def test_instantiate(self, ssh):
- mock_ssh(ssh)
-
+ def test_instantiate(self, ssh, *args):
+ test_base.mock_ssh(ssh)
nodes = {
'vnf1': 'name1',
'vnf2': 'name2',
}
- context1 = mock.Mock()
- context1._get_server.return_value = None
- context2 = mock.Mock()
- context2._get_server.return_value = context2
-
- try:
- Context.list.clear()
- except AttributeError:
- # clear() but works in Py2.7
- Context.list[:] = []
-
- Context.list.extend([
- context1,
- context2,
- ])
-
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf.scenario_helper.scenario_cfg = {
+ 'nodes': {sample_vnf.name: 'mock'}
+ }
sample_vnf.APP_NAME = 'sample1'
sample_vnf._start_server = mock.Mock(return_value=0)
sample_vnf._vnf_process = mock.MagicMock()
}
self.assertIsNone(sample_vnf.instantiate(scenario_cfg, {}))
- self.assertEqual(sample_vnf.nfvi_context, context2)
def test__update_collectd_options(self):
scenario_cfg = {'options':
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time")
@mock.patch("yardstick.ssh.SSH")
def test_wait_for_instantiate_empty_queue(self, ssh, *args):
- mock_ssh(ssh, exec_result=(1, "", ""))
+ test_base.mock_ssh(ssh, exec_result=(1, "", ""))
queue_size_list = [
0,
self.assertEqual(sample_vnf.get_stats(), 'the stats')
- def test_collect_kpi(self):
+ @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)
+ sample_vnf.scenario_helper.scenario_cfg = {
+ 'nodes': {sample_vnf.name: "mock"}
+ }
sample_vnf.APP_NAME = 'sample1'
sample_vnf.COLLECT_KPI = r'\s(\d+)\D*(\d+)\D*(\d+)'
sample_vnf.COLLECT_MAP = {
'k2': 34,
'k3': 91,
'collect_stats': {},
+ 'physical_node': 'mock_node'
}
result = sample_vnf.collect_kpi()
self.assertDictEqual(result, expected)
- def test_collect_kpi_default(self):
+ @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)
+ sample_vnf.scenario_helper.scenario_cfg = {
+ 'nodes': {sample_vnf.name: "mock"}
+ }
sample_vnf.APP_NAME = 'sample1'
sample_vnf.COLLECT_KPI = r'\s(\d+)\D*(\d+)\D*(\d+)'
sample_vnf.get_stats = mock.Mock(return_value='')
expected = {
+ 'physical_node': 'mock_node',
'packets_in': 0,
'packets_fwd': 0,
'packets_dropped': 0,