# 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.
-#
-# Unittest for yardstick.network_services.traffic_profile.test_base
+import sys
-from __future__ import absolute_import
-import unittest
import mock
+import unittest
-from yardstick.network_services.traffic_profile.base import \
- TrafficProfile, DummyProfile
+from yardstick.common import exceptions
+from yardstick.network_services import traffic_profile as tprofile_package
+from yardstick.network_services.traffic_profile import base
+from yardstick.tests import unit as unit_test
class TestTrafficProfile(unittest.TestCase):
return _mock
def test___init__(self):
- traffic_profile = TrafficProfile(self.TRAFFIC_PROFILE)
+ traffic_profile = base.TrafficProfile(self.TRAFFIC_PROFILE)
self.assertEqual(self.TRAFFIC_PROFILE, traffic_profile.params)
def test_execute(self):
- traffic_profile = TrafficProfile(self.TRAFFIC_PROFILE)
- self.assertRaises(NotImplementedError, traffic_profile.execute_traffic, {})
+ traffic_profile = base.TrafficProfile(self.TRAFFIC_PROFILE)
+ self.assertRaises(NotImplementedError,
+ traffic_profile.execute_traffic, {})
+
+ def test_get_existing_traffic_profile(self):
+ traffic_profile_list = [
+ 'RFC2544Profile', 'FixedProfile', 'TrafficProfileGenericHTTP',
+ 'IXIARFC2544Profile', 'ProxACLProfile', 'ProxBinSearchProfile',
+ 'ProxProfile', 'ProxRampProfile']
+ with mock.patch.dict(sys.modules, unit_test.STL_MOCKS):
+ tprofile_package.register_modules()
+
+ for tp in traffic_profile_list:
+ traffic_profile = base.TrafficProfile.get(
+ {'traffic_profile': {'traffic_type': tp}})
+ self.assertEqual(tp, traffic_profile.__class__.__name__)
- def test_get(self):
- traffic_profile = TrafficProfile(self.TRAFFIC_PROFILE)
- self.assertRaises(RuntimeError, traffic_profile.get,
- self.TRAFFIC_PROFILE)
+ def test_get_non_existing_traffic_profile(self):
+ self.assertRaises(exceptions.TrafficProfileNotImplemented,
+ base.TrafficProfile.get, self.TRAFFIC_PROFILE)
class TestDummyProfile(unittest.TestCase):
def test_execute(self):
- dummy_profile = DummyProfile(TrafficProfile)
+ dummy_profile = base.DummyProfile(base.TrafficProfile)
self.assertIsNone(dummy_profile.execute({}))
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()
acl_approx_vnf.ssh_helper.run.assert_called_once()
@mock.patch("yardstick.network_services.vnf_generic.vnf.acl_vnf.YangModel")
- @mock.patch("yardstick.network_services.vnf_generic.vnf.acl_vnf.find_relative_file")
+ @mock.patch.object(utils, 'find_relative_file')
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.Context")
@mock.patch(SSH_HELPER)
def test_instantiate(self, ssh, *args):
import unittest
from tests.unit import STL_MOCKS
+from yardstick.common import utils
from yardstick.network_services.vnf_generic.vnf.base import VnfdHelper
result = setup_helper.prox_config_data
self.assertEqual(result, expected)
- @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.find_relative_file')
+ @mock.patch.object(utils, 'find_relative_file')
def test_build_config_file_no_additional_file(self, mock_find_path):
vnf1 = {
'prox_args': {'-c': ""},
self.assertEqual(helper._prox_config_data, '4')
self.assertEqual(helper.remote_path, '5')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.find_relative_file')
+ @mock.patch.object(utils, 'find_relative_file')
def test_build_config_file_additional_file_string(self, mock_find_path):
vnf1 = {
'prox_args': {'-c': ""},
helper.build_config_file()
self.assertDictEqual(helper.additional_files, expected)
- @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.find_relative_file')
+ @mock.patch.object(utils, 'find_relative_file')
def test_build_config_file_additional_file(self, mock_find_path):
vnf1 = {
'prox_args': {'-c': ""},
mock_meminfo.assert_called_once_with(ssh_helper)
@mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.open')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.find_relative_file')
+ @mock.patch.object(utils, 'find_relative_file')
@mock.patch('yardstick.network_services.vnf_generic.vnf.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()
-#!/usr/bin/env python
-
# Copyright (c) 2016-2017 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# limitations under the License.
#
-from __future__ import absolute_import
+import subprocess
-import unittest
import mock
-import subprocess
+import unittest
+import six
from tests.unit import STL_MOCKS
+from yardstick import ssh
+from yardstick.common import utils
STLClient = mock.MagicMock()
ixload_traffic_gen = IxLoadTrafficGen(NAME, vnfd)
self.assertEqual(None, ixload_traffic_gen.listen_traffic({}))
- @mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.makedirs")
+ @mock.patch.object(utils, 'find_relative_file')
+ @mock.patch.object(utils, 'makedirs')
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.call")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.shutil")
- def test_instantiate(self, call, shutil, mock_makedirs):
- # pylint: disable=unused-argument
+ def test_instantiate(self, shutil, *args):
with mock.patch("yardstick.ssh.SSH") as ssh:
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
'1C/1T',
'worker_threads': 1}}
}})
- with mock.patch('yardstick.benchmark.scenarios.networking.vnf_generic.open',
- create=True) as mock_open:
+ with mock.patch.object(six.moves.builtins, 'open',
+ create=True) as mock_open:
mock_open.return_value = mock.MagicMock()
ixload_traffic_gen.instantiate(scenario_cfg, {})
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.call")
- @mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.shutil")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.open")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.min")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.max")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.len")
- def test_run_traffic(self, call, shutil, main_open, min, max, len):
- # pylint: disable=unused-argument
+ @mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.shutil")
+ def test_run_traffic(self, shutil, *args):
mock_traffic_profile = mock.Mock(autospec=TrafficProfile)
mock_traffic_profile.get_traffic_definition.return_value = "64"
mock_traffic_profile.params = self.TRAFFIC_PROFILE
self.assertIsNone(result)
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.call")
- @mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.shutil")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.open")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.min")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.max")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.len")
- def test_run_traffic_csv(self, call, shutil, main_open, min, max, len):
- # pylint: disable=unused-argument
+ @mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.shutil")
+ def test_run_traffic_csv(self, shutil, *args):
mock_traffic_profile = mock.Mock(autospec=TrafficProfile)
mock_traffic_profile.get_traffic_definition.return_value = "64"
mock_traffic_profile.params = self.TRAFFIC_PROFILE
self.assertIsNone(result)
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.call")
+ @mock.patch.object(ssh, 'SSH')
def test_terminate(self, *args):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh.from_node.return_value = ssh_mock
- ixload_traffic_gen = IxLoadTrafficGen(NAME, vnfd)
- self.assertEqual(None, ixload_traffic_gen.terminate())
+ vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
+ ixload_traffic_gen = IxLoadTrafficGen(NAME, vnfd)
+ self.assertEqual(None, ixload_traffic_gen.terminate())
- @mock.patch("yardstick.ssh.SSH")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.call")
- def test_parse_csv_read(self, mock_call, mock_ssh):
- # pylint: disable=unused-argument
+ @mock.patch.object(ssh, 'SSH')
+ def test_parse_csv_read(self, mock_ssh, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
kpi_data = {
'HTTP Total Throughput (Kbps)': 1,
for key_left, key_right in IxLoadResourceHelper.KPI_LIST.items():
self.assertEqual(result[key_left][-1], int(kpi_data[key_right]))
- @mock.patch("yardstick.ssh.SSH")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.call")
- def test_parse_csv_read_value_error(self, mock_call, mock_ssh):
- # pylint: disable=unused-argument
+ @mock.patch.object(ssh, 'SSH')
+ def test_parse_csv_read_value_error(self, mock_ssh, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
http_reader = [{
'HTTP Total Throughput (Kbps)': 1,
ixload_traffic_gen.resource_helper.parse_csv_read(http_reader)
self.assertDictEqual(ixload_traffic_gen.resource_helper.result, init_value)
- @mock.patch("yardstick.ssh.SSH")
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_ixload.call")
- def test_parse_csv_read_error(self, mock_call, mock_ssh):
- # pylint: disable=unused-argument
+ @mock.patch.object(ssh, 'SSH')
+ def test_parse_csv_read_error(self, mock_ssh, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
http_reader = [{
'HTTP Total Throughput (Kbps)': 1,
#
import os
-import unittest
+
import mock
+import six
+import unittest
from tests.unit import STL_MOCKS
'task_path': '/path/to/task'
}
- @mock.patch('yardstick.benchmark.scenarios.networking.vnf_generic.open', create=True)
+ @mock.patch.object(six.moves.builtins, 'open', create=True)
@mock.patch('yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.open',
mock.mock_open(), create=True)
@mock.patch('yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.LOG.exception')
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()
stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
stl_patch.start()
vfw_approx_vnf._run()
vfw_approx_vnf.ssh_helper.run.assert_called_once()
- @mock.patch("yardstick.network_services.vnf_generic.vnf.vfw_vnf.find_relative_file")
+ @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.sample_vnf.Context")
@mock.patch(SSH_HELPER)
# 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.
-""" NSPerf specific scenario definition """
-
-from __future__ import absolute_import
+from collections import defaultdict
+import copy
import logging
-import errno
-
import ipaddress
-
-import copy
+from itertools import chain
import os
-import sys
import re
-from itertools import chain
+import sys
import six
import yaml
-from collections import defaultdict
-from yardstick.benchmark.scenarios import base
+from yardstick.benchmark.scenarios import base as scenario_base
from yardstick.common.constants import LOG_DIR
from yardstick.common.process import terminate_children
-from yardstick.common.utils import import_modules_from_package, itersubclasses
+from yardstick.common import utils
from yardstick.common.yaml_loader import yaml_load
from yardstick.network_services.collector.subscriber import Collector
from yardstick.network_services.vnf_generic import vnfdgen
from yardstick.network_services.vnf_generic.vnf.base import GenericVNF
-from yardstick.network_services.traffic_profile.base import TrafficProfile
+from yardstick.network_services import traffic_profile
+from yardstick.network_services.traffic_profile import base as tprofile_base
from yardstick.network_services.utils import get_nsb_option
from yardstick import ssh
+traffic_profile.register_modules()
+
+
LOG = logging.getLogger(__name__)
self.conn.close()
-def find_relative_file(path, task_path):
- """
- Find file in one of places: in abs of path or
- relative to TC scenario file. In this order.
-
- :param path:
- :param task_path:
- :return str: full path to file
- """
- # fixme: create schema to validate all fields have been provided
- for lookup in [os.path.abspath(path), os.path.join(task_path, path)]:
- try:
- with open(lookup):
- return lookup
- except IOError:
- pass
- raise IOError(errno.ENOENT, 'Unable to find {} file'.format(path))
-
-
-def open_relative_file(path, task_path):
- try:
- return open(path)
- except IOError as e:
- if e.errno == errno.ENOENT:
- return open(os.path.join(task_path, path))
- raise
-
-
-class NetworkServiceTestCase(base.Scenario):
+class NetworkServiceTestCase(scenario_base.Scenario):
"""Class handles Generic framework to do pre-deployment VNF &
Network service testing """
self.context_cfg = context_cfg
# fixme: create schema to validate all fields have been provided
- with open_relative_file(scenario_cfg["topology"],
- scenario_cfg['task_path']) as stream:
+ with utils.open_relative_file(scenario_cfg["topology"],
+ scenario_cfg['task_path']) as stream:
topology_yaml = yaml_load(stream)
self.topology = topology_yaml["nsd:nsd-catalog"]["nsd"][0]
def _get_traffic_profile(self):
profile = self.scenario_cfg["traffic_profile"]
path = self.scenario_cfg["task_path"]
- with open_relative_file(profile, path) as infile:
+ with utils.open_relative_file(profile, path) as infile:
return infile.read()
def _fill_traffic_profile(self):
traffic_map_data = {
'flow': self._get_traffic_flow(),
'imix': self._get_traffic_imix(),
- TrafficProfile.UPLINK: {},
- TrafficProfile.DOWNLINK: {},
+ tprofile_base.TrafficProfile.UPLINK: {},
+ tprofile_base.TrafficProfile.DOWNLINK: {},
}
traffic_vnfd = vnfdgen.generate_vnfd(traffic_mapping, traffic_map_data)
- self.traffic_profile = TrafficProfile.get(traffic_vnfd)
+ self.traffic_profile = tprofile_base.TrafficProfile.get(traffic_vnfd)
return self.traffic_profile
def _find_vnf_name_from_id(self, vnf_id):
self._resolve_topology()
self._update_context_with_topology()
- FIND_NETDEVICE_STRING = r"""find /sys/devices/pci* -type d -name net -exec sh -c '{ grep -sH ^ \
-$1/ifindex $1/address $1/operstate $1/device/vendor $1/device/device \
-$1/device/subsystem_vendor $1/device/subsystem_device ; \
-printf "%s/driver:" $1 ; basename $(readlink -s $1/device/driver); } \
-' sh \{\}/* \;
-"""
+ FIND_NETDEVICE_STRING = (
+ r"""find /sys/devices/pci* -type d -name net -exec sh -c '{ grep -sH ^ \
+ $1/ifindex $1/address $1/operstate $1/device/vendor $1/device/device \
+ $1/device/subsystem_vendor $1/device/subsystem_device ; \
+ printf "%s/driver:" $1 ; basename $(readlink -s $1/device/driver); } \
+ ' sh \{\}/* \;
+ """)
BASE_ADAPTER_RE = re.compile(
'^/sys/devices/(.*)/net/([^/]*)/([^:]*):(.*)$', re.M)
:param vnf_model_id: parsed vnfd model ID field
:return: subclass of GenericVNF
"""
- import_modules_from_package(
+ utils.import_modules_from_package(
"yardstick.network_services.vnf_generic.vnf")
expected_name = vnf_model_id
classes_found = []
def impl():
- for name, class_ in ((c.__name__, c) for c in itersubclasses(GenericVNF)):
+ for name, class_ in ((c.__name__, c) for c in
+ utils.itersubclasses(GenericVNF)):
if name == expected_name:
yield class_
classes_found.append(name)
LOG.debug("no model for %s, skipping", node_name)
continue
file_path = scenario_cfg['task_path']
- with open_relative_file(file_name, file_path) as stream:
+ with utils.open_relative_file(file_name, file_path) as stream:
vnf_model = stream.read()
vnfd = vnfdgen.generate_vnfd(vnf_model, node)
# TODO: here add extra context_cfg["nodes"] regardless of template
class IPv6RangeError(YardstickException):
message = 'Start IP "%(start_ip)s" is greater than end IP "%(end_ip)s"'
+
+class TrafficProfileNotImplemented(YardstickException):
+ message = 'No implementation for traffic profile %(profile_class)s.'
+
+
class DPDKSetupDriverError(YardstickException):
message = '"igb_uio" driver is not loaded'
yield sub
-def import_modules_from_package(package):
+def import_modules_from_package(package, raise_exception=False):
"""Import modules given a package name
:param: package - Full package name. For example: rally.deploy.engines
for module_name in missing_modules:
try:
importlib.import_module(module_name)
- except (ImportError, SyntaxError):
+ except (ImportError, SyntaxError) as exc:
+ if raise_exception:
+ raise exc
logger.exception('Unable to import module %s', module_name)
output[match[0]] = match[1]
return output
+
+
+def find_relative_file(path, task_path):
+ """
+ Find file in one of places: in abs of path or relative to a directory path,
+ in this order.
+
+ :param path:
+ :param task_path:
+ :return str: full path to file
+ """
+ # fixme: create schema to validate all fields have been provided
+ for lookup in [os.path.abspath(path), os.path.join(task_path, path)]:
+ try:
+ with open(lookup):
+ return lookup
+ except IOError:
+ pass
+ raise IOError(errno.ENOENT, 'Unable to find {} file'.format(path))
+
+
+def open_relative_file(path, task_path):
+ try:
+ return open(path)
+ except IOError as e:
+ if e.errno == errno.ENOENT:
+ return open(os.path.join(task_path, path))
+ raise
+# Copyright (c) 2018 Intel Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+
+import importlib
+
+
+def register_modules():
+ modules = [
+ 'yardstick.network_services.traffic_profile.trex_traffic_profile',
+ 'yardstick.network_services.traffic_profile.fixed',
+ 'yardstick.network_services.traffic_profile.http',
+ 'yardstick.network_services.traffic_profile.http_ixload',
+ 'yardstick.network_services.traffic_profile.ixia_rfc2544',
+ 'yardstick.network_services.traffic_profile.prox_ACL',
+ 'yardstick.network_services.traffic_profile.prox_binsearch',
+ 'yardstick.network_services.traffic_profile.prox_profile',
+ 'yardstick.network_services.traffic_profile.prox_ramp',
+ 'yardstick.network_services.traffic_profile.rfc2544',
+ ]
+
+ for module in modules:
+ importlib.import_module(module)
# 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.
-""" Base class for the generic traffic profile implementation """
-from __future__ import absolute_import
-from yardstick.common.utils import import_modules_from_package, itersubclasses
+from yardstick.common import exceptions
+from yardstick.common import utils
class TrafficProfile(object):
:return:
"""
profile_class = tp_config["traffic_profile"]["traffic_type"]
- import_modules_from_package(
- "yardstick.network_services.traffic_profile")
try:
- return next(c for c in itersubclasses(TrafficProfile)
+ return next(c for c in utils.itersubclasses(TrafficProfile)
if c.__name__ == profile_class)(tp_config)
except StopIteration:
- raise RuntimeError("No implementation for %s", profile_class)
+ raise exceptions.TrafficProfileNotImplemented(
+ profile_class=profile_class)
def __init__(self, tp_config):
# e.g. RFC2544 start_ip, stop_ip, drop_rate,
import six
from yardstick.common import exceptions as y_exc
-from yardstick.network_services.traffic_profile.base import TrafficProfile
+from yardstick.network_services.traffic_profile import base
from trex_stl_lib.trex_stl_client import STLStream
from trex_stl_lib.trex_stl_streams import STLFlowLatencyStats
from trex_stl_lib.trex_stl_streams import STLTXCont
LOG = logging.getLogger(__name__)
-class TrexProfile(TrafficProfile):
+class TrexProfile(base.TrafficProfile):
""" This class handles Trex Traffic profile generation and execution """
PROTO_MAP = {
self.vm_flow_vars.append(stl_vm_wr_flow_var)
return partial
- def _dscp_range_action_partial(self, *_):
+ def _dscp_range_action_partial(self, *args):
def partial(min_value, max_value, count):
# pylint: disable=unused-argument
stl_vm_flow_var = STLVmFlowVar(name="dscp",
# See the License for the specific language governing permissions and
# limitations under the License.
-from __future__ import absolute_import
-from __future__ import print_function
import logging
-from yardstick.benchmark.scenarios.networking.vnf_generic import find_relative_file
+from yardstick.common import utils
from yardstick.network_services.vnf_generic.vnf.sample_vnf import SampleVNF, DpdkVnfSetupEnvHelper
from yardstick.network_services.yang_model import YangModel
self.acl_rules = None
def _start_vnf(self):
- yang_model_path = find_relative_file(self.scenario_helper.options['rules'],
- self.scenario_helper.task_path)
+ yang_model_path = utils.find_relative_file(
+ self.scenario_helper.options['rules'],
+ self.scenario_helper.task_path)
yang_model = YangModel(yang_model_path)
self.acl_rules = yang_model.get_rules()
super(AclApproxVnf, self)._start_vnf()
from six.moves import cStringIO
from six.moves import zip, StringIO
-from yardstick.benchmark.scenarios.networking.vnf_generic import find_relative_file
from yardstick.common import utils
from yardstick.common.utils import SocketTopology, join_non_strings, try_int
from yardstick.network_services.helpers.iniparser import ConfigParser
options = self.scenario_helper.options
config_path = options['prox_config']
config_file = os.path.basename(config_path)
- config_path = find_relative_file(config_path, task_path)
+ config_path = utils.find_relative_file(config_path, task_path)
self.additional_files = {}
try:
prox_files = [prox_files]
for key_prox_file in prox_files:
base_prox_file = os.path.basename(key_prox_file)
- key_prox_path = find_relative_file(key_prox_file, task_path)
+ key_prox_path = utils.find_relative_file(key_prox_file, task_path)
remote_prox_file = self.copy_to_target(key_prox_path, base_prox_file)
self.additional_files[base_prox_file] = remote_prox_file
from trex_stl_lib.trex_stl_client import STLClient
from trex_stl_lib.trex_stl_exceptions import STLError
from yardstick.benchmark.contexts.base import Context
-from yardstick.benchmark.scenarios.networking.vnf_generic import find_relative_file
from yardstick.common import exceptions as y_exceptions
-from yardstick.common import utils
from yardstick.common.process import check_if_process_failed
+from yardstick.common import utils
from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkBindHelper
from yardstick.network_services.helpers.samplevnf_helper import PortPairs
from yardstick.network_services.helpers.samplevnf_helper import MultiPortConfig
'vnf_type': self.VNF_TYPE,
}
- config_tpl_cfg = find_relative_file(self.DEFAULT_CONFIG_TPL_CFG, task_path)
+ config_tpl_cfg = utils.find_relative_file(self.DEFAULT_CONFIG_TPL_CFG,
+ task_path)
config_basename = posixpath.basename(self.CFG_CONFIG)
script_basename = posixpath.basename(self.CFG_SCRIPT)
multiport = MultiPortConfig(self.scenario_helper.topology,
from collections import OrderedDict
from subprocess import call
-from yardstick.common.utils import makedirs
+from yardstick.common import utils
from yardstick.network_services.vnf_generic.vnf.sample_vnf import SampleVNFTrafficGen
from yardstick.network_services.vnf_generic.vnf.sample_vnf import ClientResourceHelper
-from yardstick.benchmark.scenarios.networking.vnf_generic import find_relative_file
+
LOG = logging.getLogger(__name__)
def setup(self):
# NOTE: fixup scenario_helper to hanlde ixia
self.resource_file_name = \
- find_relative_file(self.scenario_helper.scenario_cfg['ixia_profile'],
- self.scenario_helper.scenario_cfg["task_path"])
- makedirs(self.RESULTS_MOUNT)
+ utils.find_relative_file(
+ self.scenario_helper.scenario_cfg['ixia_profile'],
+ self.scenario_helper.scenario_cfg["task_path"])
+ utils.makedirs(self.RESULTS_MOUNT)
cmd = MOUNT_CMD.format(self.vnfd_helper.mgmt_interface, self)
LOG.debug(cmd)
call(cmd, shell=True)
shutil.rmtree(self.RESULTS_MOUNT, ignore_errors=True)
- makedirs(self.RESULTS_MOUNT)
+ utils.makedirs(self.RESULTS_MOUNT)
shutil.copy(self.resource_file_name, self.RESULTS_MOUNT)
def make_aggregates(self):
import logging
import sys
-from yardstick.common.utils import ErrorClass
+from yardstick.common import utils
from yardstick.network_services.vnf_generic.vnf.sample_vnf import SampleVNFTrafficGen
from yardstick.network_services.vnf_generic.vnf.sample_vnf import ClientResourceHelper
from yardstick.network_services.vnf_generic.vnf.sample_vnf import Rfc2544ResourceHelper
-from yardstick.benchmark.scenarios.networking.vnf_generic import find_relative_file
+
LOG = logging.getLogger(__name__)
try:
from IxNet import IxNextgen
except ImportError:
- IxNextgen = ErrorClass
+ IxNextgen = utils.ErrorClass
class IxiaRfc2544Helper(Rfc2544ResourceHelper):
# we don't know client_file_name until runtime as instantiate
client_file_name = \
- find_relative_file(self.scenario_helper.scenario_cfg['ixia_profile'],
- self.scenario_helper.scenario_cfg["task_path"])
+ utils.find_relative_file(
+ self.scenario_helper.scenario_cfg['ixia_profile'],
+ self.scenario_helper.scenario_cfg["task_path"])
self.client.ix_load_config(client_file_name)
time.sleep(WAIT_AFTER_CFG_LOAD)
# See the License for the specific language governing permissions and
# limitations under the License.
-from __future__ import absolute_import
import logging
-from yardstick.benchmark.scenarios.networking.vnf_generic import find_relative_file
+from yardstick.common import utils
from yardstick.network_services.vnf_generic.vnf.sample_vnf import SampleVNF, DpdkVnfSetupEnvHelper
from yardstick.network_services.yang_model import YangModel
self.vfw_rules = None
def _start_vnf(self):
- yang_model_path = find_relative_file(self.scenario_helper.options['rules'],
- self.scenario_helper.task_path)
+ yang_model_path = utils.find_relative_file(
+ self.scenario_helper.options['rules'],
+ self.scenario_helper.task_path)
yang_model = YangModel(yang_model_path)
self.vfw_rules = yang_model.get_rules()
super(FWApproxVnf, self)._start_vnf()
--- /dev/null
+# Copyright 2018 Intel Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+
+import abc
+
+import six
+import unittest
+
+
+@six.add_metaclass(abc.ABCMeta)
+class BaseUnitTestCase(unittest.TestCase):
+ """Base class for unit tests"""
# 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.
-#
-import os
+from copy import deepcopy
import errno
-import unittest
-import mock
+import os
-from copy import deepcopy
+import mock
+import six
+import unittest
-from yardstick.benchmark.scenarios.networking.vnf_generic import \
- SshManager, NetworkServiceTestCase, IncorrectConfig, \
- open_relative_file
+from yardstick.common import utils
+from yardstick.benchmark.scenarios.networking import vnf_generic
from yardstick.network_services.collector.subscriber import Collector
from yardstick.network_services.vnf_generic.vnf.base import \
GenericTrafficGen, GenericVNF
},
}
- self.s = NetworkServiceTestCase(self.scenario_cfg, self.context_cfg)
+ self.s = vnf_generic.NetworkServiceTestCase(self.scenario_cfg,
+ self.context_cfg)
def _get_file_abspath(self, filename):
curr_path = os.path.dirname(os.path.abspath(__file__))
ssh_mock.execute = \
mock.Mock(return_value=(0, SYS_CLASS_NET + IP_ADDR_SHOW, ""))
ssh.from_node.return_value = ssh_mock
- for _, node_dict in self.context_cfg["nodes"].items():
- with SshManager(node_dict) as conn:
+ for node_dict in self.context_cfg["nodes"].values():
+ with vnf_generic.SshManager(node_dict) as conn:
self.assertIsNotNone(conn)
def test___init__(self):
},
}
+ # NOTE(ralonsoh): this variable is not used in the test case and maybe
+ # it should.
+ #result = {'flow': {'dst_ip0': '152.16.40.2-152.16.40.254',
+ # 'src_ip0': '152.16.100.2-152.16.100.254'}}
self.assertEqual({'flow': {}}, self.s._get_traffic_flow())
def test___get_traffic_flow_error(self):
vnfd = COMPLETE_TREX_VNFD['vnfd:vnfd-catalog']['vnfd'][0]['class-name']
self.assertIsNotNone(self.s.get_vnf_impl(vnfd))
- with self.assertRaises(IncorrectConfig) as raised:
+ with self.assertRaises(vnf_generic.IncorrectConfig) as raised:
self.s.get_vnf_impl('NonExistentClass')
exc_str = str(raised.exception)
mock.Mock(return_value=(1, SYS_CLASS_NET + IP_ADDR_SHOW, ""))
ssh.from_node.return_value = ssh_mock
- with self.assertRaises(IncorrectConfig):
+ with self.assertRaises(vnf_generic.IncorrectConfig):
self.s.map_topology_to_infrastructure()
def test_map_topology_to_infrastructure_config_invalid(self):
mock.Mock(return_value=(0, SYS_CLASS_NET + IP_ADDR_SHOW, ""))
ssh.from_node.return_value = ssh_mock
- with self.assertRaises(IncorrectConfig):
+ with self.assertRaises(vnf_generic.IncorrectConfig):
self.s.map_topology_to_infrastructure()
def test__resolve_topology_invalid_config(self):
del interface['local_mac']
with mock.patch(
- "yardstick.benchmark.scenarios.networking.vnf_generic.LOG"):
- with self.assertRaises(IncorrectConfig) as raised:
+ "yardstick.benchmark.scenarios.networking.vnf_generic.LOG"):
+ with self.assertRaises(vnf_generic.IncorrectConfig) as raised:
self.s._resolve_topology()
self.assertIn('not found', str(raised.exception))
self.s.topology["vld"][0]['vnfd-connection-point-ref'].append(
self.s.topology["vld"][0]['vnfd-connection-point-ref'][0])
- with mock.patch(
- "yardstick.benchmark.scenarios.networking.vnf_generic.LOG"):
- with self.assertRaises(IncorrectConfig) as raised:
+ with mock.patch.object(vnf_generic, 'LOG'):
+ with self.assertRaises(vnf_generic.IncorrectConfig) as raised:
self.s._resolve_topology()
self.assertIn('wrong endpoint count', str(raised.exception))
self.s.topology["vld"][0]['vnfd-connection-point-ref'] = \
self.s.topology["vld"][0]['vnfd-connection-point-ref'][:1]
- with mock.patch(
- "yardstick.benchmark.scenarios.networking.vnf_generic.LOG"):
- with self.assertRaises(IncorrectConfig) as raised:
+ with mock.patch.object(vnf_generic, 'LOG'):
+ with self.assertRaises(vnf_generic.IncorrectConfig) as raised:
self.s._resolve_topology()
self.assertIn('wrong endpoint count', str(raised.exception))
/sys/devices/pci0000:00/0000:00:19.0/net/lan/driver:e1000e
/sys/devices/pci0000:00/0000:00:19.0/net/lan/pci_bus_id:0000:00:19.0
"""
- res = NetworkServiceTestCase.parse_netdev_info(output)
+ res = vnf_generic.NetworkServiceTestCase.parse_netdev_info(output)
assert res == self.SAMPLE_NETDEVS
def test_parse_netdev_info_virtio(self):
/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/device/device:0x0001
/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/driver:virtio_net
"""
- res = NetworkServiceTestCase.parse_netdev_info(output)
+ res = vnf_generic.NetworkServiceTestCase.parse_netdev_info(output)
assert res == self.SAMPLE_VM_NETDEVS
def test_probe_missing_values(self):
netdevs = self.SAMPLE_NETDEVS.copy()
network = {'local_mac': '0a:de:ad:be:ef:f5'}
- NetworkServiceTestCase._probe_missing_values(netdevs, network)
+ vnf_generic.NetworkServiceTestCase._probe_missing_values(netdevs,
+ network)
assert network['vpci'] == '0000:0b:00.0'
network = {'local_mac': '0a:de:ad:be:ef:f4'}
- NetworkServiceTestCase._probe_missing_values(netdevs, network)
+ vnf_generic.NetworkServiceTestCase._probe_missing_values(netdevs,
+ network)
assert network['vpci'] == '0000:00:19.0'
- # TODO: Split this into several tests, for different IOError sub-types
- def test_open_relative_path(self):
- mock_open = mock.mock_open()
+ @mock.patch.object(six.moves.builtins, 'open')
+ def test_open_relative_path(self, mock_open):
+ # NOTE(ralonsoh): the mocked function is not properly used and tested.
mock_open_result = mock_open()
mock_open_call_count = 1 # initial call to get result
-
- module_name = \
- 'yardstick.benchmark.scenarios.networking.vnf_generic.open'
-
- # test
- with mock.patch(module_name, mock_open, create=True):
- self.assertEqual(open_relative_file(
- 'foo', 'bar'), mock_open_result)
-
- mock_open_call_count += 1 # one more call expected
- self.assertEqual(mock_open.call_count, mock_open_call_count)
- self.assertIn('foo', mock_open.call_args_list[-1][0][0])
- self.assertNotIn('bar', mock_open.call_args_list[-1][0][0])
-
- def open_effect(*args, **kwargs):
- if kwargs.get('name', args[0]) == os.path.join('bar', 'foo'):
- return mock_open_result
- raise IOError(errno.ENOENT, 'not found')
-
- mock_open.side_effect = open_effect
- self.assertEqual(open_relative_file(
- 'foo', 'bar'), mock_open_result)
-
- mock_open_call_count += 2 # two more calls expected
- self.assertEqual(mock_open.call_count, mock_open_call_count)
- self.assertIn('foo', mock_open.call_args_list[-1][0][0])
- self.assertIn('bar', mock_open.call_args_list[-1][0][0])
-
- # test an IOError of type ENOENT
- mock_open.side_effect = IOError(errno.ENOENT, 'not found')
- with self.assertRaises(IOError):
- # the second call still raises
- open_relative_file('foo', 'bar')
-
- mock_open_call_count += 2 # two more calls expected
- self.assertEqual(mock_open.call_count, mock_open_call_count)
- self.assertIn('foo', mock_open.call_args_list[-1][0][0])
- self.assertIn('bar', mock_open.call_args_list[-1][0][0])
-
- # test an IOError other than ENOENT
- mock_open.side_effect = IOError(errno.EBUSY, 'busy')
- with self.assertRaises(IOError):
- open_relative_file('foo', 'bar')
-
- mock_open_call_count += 1 # one more call expected
- self.assertEqual(mock_open.call_count, mock_open_call_count)
+ self.assertEqual(utils.open_relative_file('foo', 'bar'),
+ mock_open_result)
+
+ mock_open_call_count += 1 # one more call expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
+ self.assertIn('foo', mock_open.call_args_list[-1][0][0])
+ self.assertNotIn('bar', mock_open.call_args_list[-1][0][0])
+
+ def open_effect(*args, **kwargs):
+ if kwargs.get('name', args[0]) == os.path.join('bar', 'foo'):
+ return mock_open_result
+ raise IOError(errno.ENOENT, 'not found')
+
+ mock_open.side_effect = open_effect
+ self.assertEqual(utils.open_relative_file('foo', 'bar'),
+ mock_open_result)
+
+ mock_open_call_count += 2 # two more calls expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
+ self.assertIn('foo', mock_open.call_args_list[-1][0][0])
+ self.assertIn('bar', mock_open.call_args_list[-1][0][0])
+
+ # test an IOError of type ENOENT
+ mock_open.side_effect = IOError(errno.ENOENT, 'not found')
+ with self.assertRaises(IOError):
+ # the second call still raises
+ utils.open_relative_file('foo', 'bar')
+
+ mock_open_call_count += 2 # two more calls expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
+ self.assertIn('foo', mock_open.call_args_list[-1][0][0])
+ self.assertIn('bar', mock_open.call_args_list[-1][0][0])
+
+ # test an IOError other than ENOENT
+ mock_open.side_effect = IOError(errno.EBUSY, 'busy')
+ with self.assertRaises(IOError):
+ utils.open_relative_file('foo', 'bar')
+
+ mock_open_call_count += 1 # one more call expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
# License for the specific language governing permissions and limitations
# under the License.
-import unittest
-
from yardstick.benchmark.scenarios import base
+from yardstick.tests.unit import base as ut_base
-class ScenarioTestCase(unittest.TestCase):
+class ScenarioTestCase(ut_base.BaseUnitTestCase):
def test_get_scenario_type(self):
scenario_type = 'dummy scenario'
str(exc.exception))
-class IterScenarioClassesTestCase(unittest.TestCase):
+class IterScenarioClassesTestCase(ut_base.BaseUnitTestCase):
def test_no_scenario_type_defined(self):
some_existing_scenario_class_names = [
Code Review / yardstick.git / commitdiff