- role: install_samplevnf
vnf_name: FW
- role: install_samplevnf
- vnf_name: CGNATP
+ vnf_name: CGNAPT
# build shared DPDK for collectd only, required DPDK downloaded already
- install_dpdk_shared
- install_rabbitmq
loop_control:
loop_var: node_item
with_items: "{{ infra_deploy_vars.nodes }}"
+
+- name: Create list of dictionaries with vm name, ip address
+ set_fact:
+ vm_name_ip: "{{ vm_name_ip|default([]) + [{item.hostname: item.interfaces[1].ip}] }}"
+ with_items: "{{ infra_deploy_vars.nodes }}"
+
+- name: Make sure VM is reachable
+ include_tasks: validate_vms.yml
+ loop_control:
+ loop_var: name_ip
+ with_items: "{{ vm_name_ip }}"
--- /dev/null
+# 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.
+---
+- name: Validate if VM is responding
+ wait_for:
+ host: "{{ item.value }}"
+ port: 22
+ timeout: 10
+ with_dict: "{{ name_ip }}"
+ register: result
+ ignore_errors: yes
+
+- name: Stop VM if it is not responding
+ virt:
+ name: "{{ item.key }}"
+ command: destroy
+ when: result is failed
+ with_dict: "{{ name_ip }}"
+
+- name: Wait for VM is being stopped
+ pause:
+ seconds: 10
+ when: result is failed
+
+- name: Start VM
+ virt:
+ name: "{{ item.key }}"
+ command: start
+ when: result is failed
+ with_dict: "{{ name_ip }}"
+
+- name: Validate if VM is responding
+ wait_for:
+ host: "{{ item.value }}"
+ port: 22
+ timeout: 10
+ when: result is failed
+ with_dict: "{{ name_ip }}"
+ register: result
+
+- fail:
+ msg: "FAILED on {{ name_ip }}"
+ when: result is failed
secondary_ip: "{{ item.interfaces[1].ip }}"
when: item.hostname == 'yardstickvm'
with_items: "{{ infra_deploy_vars.nodes }}"
-
-- name: Workaround, not all VMs are ready by that time
- pause: seconds=20
pip:
requirements: "{{ yardstick_dir }}/requirements.txt"
virtualenv: "{{ yardstick_dir }}/virtualenv"
- async: 300
+ async: 600
poll: 0
register: pip_installer
when: virtual_environment == True
- name: Install Yardstick requirements
pip:
requirements: "{{ yardstick_dir }}/requirements.txt"
- async: 300
+ async: 600
poll: 0
register: pip_installer
when: virtual_environment == False
jid: "{{ pip_installer.ansible_job_id }}"
register: job_result
until: job_result.finished
- retries: 100
+ retries: 150
- name: Install Yardstick code (venv)
pip:
--- /dev/null
+##############################################################################
+# Copyright (c) 2018 Huawei Technologies Co.,Ltd and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+
+FROM ubuntu:16.04
+
+LABEL image=opnfv/yardstick-image-k8s
+
+ARG BRANCH=master
+
+# GIT repo directory
+ENV CLONE_DEST="/opt/tempT"
+
+RUN apt-get update && apt-get install -y \
+ git bc bonnie++ fio gcc iperf3 ethtool \
+ iproute2 linux-tools-common linux-tools-generic \
+ lmbench make netperf patch perl rt-tests stress \
+ sysstat iputils-ping openssh-server sudo && \
+ apt-get -y autoremove && apt-get clean
+
+RUN rm -rf -- ${CLONE_DEST}
+RUN git clone https://github.com/kdlucas/byte-unixbench.git ${CLONE_DEST}
+RUN mkdir -p ${CLONE_DEST}/UnixBench/
+
+RUN git clone https://github.com/beefyamoeba5/ramspeed.git ${CLONE_DEST}/RAMspeed
+WORKDIR ${CLONE_DEST}/RAMspeed/ramspeed-2.6.0
+RUN mkdir -p ${CLONE_DEST}/RAMspeed/ramspeed-2.6.0/temp
+RUN bash build.sh
+
+RUN git clone https://github.com/beefyamoeba5/cachestat.git ${CLONE_DEST}/Cachestat
+
+WORKDIR /
+
+CMD /bin/bash
{% set cpu_num = cpu_num or 1 %}
{% set ram_num = ram_num or 512 %}
{% set disk_num = disk_num or 7 %}
+{% set dpdk_enabled = dpdk_enabled or False %}
scenarios:
{% for num in range(stack_num) %}
vcpus: {{cpu_num}}
ram: {{ram_num}}
disk: {{disk_num}}
+ {% if dpdk_enabled %}
+ extra_specs:
+ hw:mem_page_size: "large"
+ {% endif %}
user: ubuntu
placement_groups:
self._delete_pod(pod)
def _delete_pod(self, pod):
- k8s_utils.delete_pod(pod)
+ k8s_utils.delete_pod(pod, skip_codes=[404])
def _create_crd(self):
LOG.info('Create Custom Resource Definition elements')
def _get_server(self, name):
node_ports = self._get_service_ports(name)
for sn_port in (sn_port for sn_port in node_ports
- if sn_port.port == constants.SSH_PORT):
- node_port = sn_port.node_port
+ if sn_port['port'] == constants.SSH_PORT):
+ node_port = sn_port['node_port']
break
else:
raise exceptions.KubernetesSSHPortNotDefined()
service = k8s_utils.get_service_by_name(service_name)
if not service:
raise exceptions.KubernetesServiceObjectNotDefined()
- return service.ports
+ ports = []
+ for port in service.ports:
+ ports.append({'name': port.name,
+ 'node_port': port.node_port,
+ 'port': port.port,
+ 'protocol': port.protocol,
+ 'target_port': port.target_port})
+ return ports
pass
self.create_interfaces_from_node(vnfd, node)
vnf_impl = self.get_vnf_impl(vnfd['id'])
- vnf_instance = vnf_impl(node_name, vnfd)
+ vnf_instance = vnf_impl(node_name, vnfd, scenario_cfg['task_id'])
vnfs.append(vnf_instance)
self.vnfs = vnfs
class YardstickBannedModuleImported(YardstickException):
- # pragma: no cover
message = 'Module "%(module)s" cannnot be imported. Reason: "%(reason)s"'
class HeatTemplateError(YardstickException):
- """Error in Heat during the stack deployment"""
message = ('Error in Heat during the creation of the OpenStack stack '
'"%(stack_name)s"')
message = 'No implementation for traffic profile %(profile_class)s.'
+class TrafficProfileRate(YardstickException):
+ message = 'Traffic profile rate must be "<number>[fps|%]"'
+
+
class DPDKSetupDriverError(YardstickException):
message = '"igb_uio" driver is not loaded'
message = 'ServiceObject is not defined'
+class KubernetesServiceObjectDefinitionError(YardstickException):
+ message = ('Kubernetes Service object definition error, missing '
+ 'parameters: %(missing_parameters)s')
+
+
+class KubernetesServiceObjectNameError(YardstickException):
+ message = ('Kubernetes Service object name "%(name)s" does not comply'
+ 'naming convention')
+
+
class KubernetesCRDObjectDefinitionError(YardstickException):
message = ('Kubernetes Custom Resource Definition Object error, missing '
'parameters: %(missing_parameters)s')
message = 'Container port not defined in "%(port)s"'
+class KubernetesContainerWrongImagePullPolicy(YardstickException):
+ message = 'Image pull policy must be "Always", "IfNotPresent" or "Never"'
+
+
+class KubernetesContainerCommandType(YardstickException):
+ message = '"args" and "command" must be string or list of strings'
+
+
class ScenarioCreateNetworkError(YardstickException):
message = 'Create Neutron Network Scenario failed'
raise
-def delete_service(name,
- namespace='default',
- **kwargs): # pragma: no cover
+def delete_service(name, namespace='default', skip_codes=None, **kwargs):
+ skip_codes = [] if not skip_codes else skip_codes
core_v1_api = get_core_api()
try:
body = client.V1DeleteOptions()
core_v1_api.delete_namespaced_service(name, namespace, body, **kwargs)
- except ApiException:
- LOG.exception('Delete Service failed')
+ except ApiException as e:
+ if e.status in skip_codes:
+ LOG.info(e.reason)
+ else:
+ raise exceptions.KubernetesApiException(
+ action='delete', resource='Service')
def get_service_list(namespace='default', **kwargs):
def delete_pod(name,
namespace='default',
wait=False,
+ skip_codes=None,
**kwargs): # pragma: no cover
# pylint: disable=unused-argument
+ skip_codes = [] if not skip_codes else skip_codes
core_v1_api = get_core_api()
body = kwargs.get('body', client.V1DeleteOptions())
kwargs.pop('body', None)
namespace,
body,
**kwargs)
- except ApiException:
- LOG.exception('Delete pod failed')
- raise
+ except ApiException as e:
+ if e.status in skip_codes:
+ LOG.info(e.reason)
+ else:
+ raise exceptions.KubernetesApiException(
+ action='delete', resource='Pod')
def read_pod(name,
action='create', resource='CustomResourceDefinition')
-def delete_custom_resource_definition(name):
+def delete_custom_resource_definition(name, skip_codes=None):
+ skip_codes = [] if not skip_codes else skip_codes
api = get_extensions_v1beta_api()
body_obj = client.V1DeleteOptions()
try:
api.delete_custom_resource_definition(name, body_obj)
- except ApiException:
- raise exceptions.KubernetesApiException(
- action='delete', resource='CustomResourceDefinition')
+ except ApiException as e:
+ if e.status in skip_codes:
+ LOG.info(e.reason)
+ else:
+ raise exceptions.KubernetesApiException(
+ action='delete', resource='CustomResourceDefinition')
def get_custom_resource_definition(kind):
"""Abstract class to define a endpoint object for a MessagingConsumer"""
def __init__(self, _id, ctx_ids, queue):
+ super(NotificationHandler, self).__init__()
self._id = _id
self._ctx_ids = ctx_ids
self._queue = queue
from yardstick.common import exceptions
from yardstick.common import utils
+from yardstick.network_services.traffic_profile import base as tp_base
log = logging.getLogger(__name__)
- Duration: in case of traffic_type="fixedDuration", amount of seconds
to inject traffic.
- Rate: in frames per seconds or percentage.
- - Type of rate: "framesPerSecond" ("bitsPerSecond" and
- "percentLineRate" no used)
+ - Type of rate: "framesPerSecond" or "percentLineRate" ("bitsPerSecond"
+ no used)
- Frame size: custom IMIX [1] definition; a list of packet size in
bytes and the weight. E.g.:
[[64, 64, 10], [128, 128, 15], [512, 512, 5]]
type = traffic_param.get('traffic_type', 'fixedDuration')
duration = traffic_param.get('duration', 30)
- rate = traffic_param['iload']
+ rate = traffic_param['rate']
+ rate_unit = (
+ 'framesPerSecond' if traffic_param['rate_unit'] ==
+ tp_base.TrafficProfileConfig.RATE_FPS else 'percentLineRate')
weighted_range_pairs = self._parse_framesize(
traffic_param['outer_l2']['framesize'])
srcmac = str(traffic_param.get('srcmac', '00:00:00:00:00:01'))
'-type', type, '-duration', duration)
self.ixnet.setMultiAttribute(
config_element + '/frameRate',
- '-rate', rate, '-type', 'framesPerSecond')
+ '-rate', rate, '-type', rate_unit)
self.ixnet.setMultiAttribute(
config_element + '/frameSize',
'-type', 'weightedPairs',
# See the License for the specific language governing permissions and
# limitations under the License.
+import re
+
from yardstick.common import exceptions
from yardstick.common import utils
This object will parse and validate the traffic profile information.
"""
-
DEFAULT_SCHEMA = 'nsb:traffic_profile:0.1'
- DEFAULT_FRAME_RATE = 100
+ DEFAULT_FRAME_RATE = '100'
DEFAULT_DURATION = 30
+ RATE_FPS = 'fps'
+ RATE_PERCENTAGE = '%'
+ RATE_REGEX = re.compile(r'([0-9]*\.[0-9]+|[0-9]+)\s*(fps|%)*(.*)')
def __init__(self, tp_config):
self.schema = tp_config.get('schema', self.DEFAULT_SCHEMA)
self.description = tp_config.get('description')
tprofile = tp_config['traffic_profile']
self.traffic_type = tprofile.get('traffic_type')
- self.frame_rate = tprofile.get('frame_rate', self.DEFAULT_FRAME_RATE)
+ self.frame_rate, self.rate_unit = self._parse_rate(
+ tprofile.get('frame_rate', self.DEFAULT_FRAME_RATE))
self.test_precision = tprofile.get('test_precision')
self.packet_sizes = tprofile.get('packet_sizes')
self.duration = tprofile.get('duration', self.DEFAULT_DURATION)
self.upper_bound = tprofile.get('upper_bound')
self.step_interval = tprofile.get('step_interval')
+ def _parse_rate(self, rate):
+ """Parse traffic profile rate
+
+ The line rate can be defined in fps or percentage over the maximum line
+ rate:
+ - frame_rate = 5000 (by default, unit is 'fps')
+ - frame_rate = 5000fps
+ - frame_rate = 25%
+
+ :param rate: (string, int) line rate in fps or %
+ :return: (tuple: int, string) line rate number and unit
+ """
+ match = self.RATE_REGEX.match(str(rate))
+ if not match:
+ exceptions.TrafficProfileRate()
+ rate = float(match.group(1))
+ unit = match.group(2) if match.group(2) else self.RATE_FPS
+ if match.group(3):
+ raise exceptions.TrafficProfileRate()
+ return rate, unit
+
class TrafficProfile(object):
"""
import logging
-from yardstick.network_services.traffic_profile.trex_traffic_profile import \
- TrexProfile
+from yardstick.network_services.traffic_profile import base as tp_base
+from yardstick.network_services.traffic_profile import trex_traffic_profile
+
LOG = logging.getLogger(__name__)
-class IXIARFC2544Profile(TrexProfile):
+class IXIARFC2544Profile(trex_traffic_profile.TrexProfile):
UPLINK = 'uplink'
DOWNLINK = 'downlink'
+ DROP_PERCENT_ROUND = 6
+ RATE_ROUND = 5
def __init__(self, yaml_data):
super(IXIARFC2544Profile, self).__init__(yaml_data)
self.rate = self.config.frame_rate
+ self.rate_unit = self.config.rate_unit
def _get_ixia_traffic_profile(self, profile_data, mac=None):
- if mac is None:
- mac = {}
-
+ mac = {} if mac is None else mac
result = {}
for traffickey, values in profile_data.items():
if not traffickey.startswith((self.UPLINK, self.DOWNLINK)):
result[traffickey] = {
'bidir': False,
- 'iload': '100',
'id': port_id,
+ 'rate': self.rate,
+ 'rate_unit': self.rate_unit,
'outer_l2': {
'framesize': value['outer_l2']['framesize'],
'framesPerSecond': True,
return result
def _ixia_traffic_generate(self, traffic, ixia_obj):
- for key, value in traffic.items():
- if key.startswith((self.UPLINK, self.DOWNLINK)):
- value['iload'] = str(self.rate)
ixia_obj.update_frame(traffic)
ixia_obj.update_ip_packet(traffic)
ixia_obj.start_traffic()
self.pg_id = 0
self.update_traffic_profile(traffic_generator)
self.max_rate = self.rate
- self.min_rate = 0
+ self.min_rate = 0.0
else:
- self.rate = round(float(self.max_rate + self.min_rate) / 2.0, 2)
+ self.rate = round(float(self.max_rate + self.min_rate) / 2.0,
+ self.RATE_ROUND)
traffic = self._get_ixia_traffic_profile(self.full_profile, mac)
self._ixia_traffic_generate(traffic, ixia_obj)
try:
drop_percent = round(
- (packet_drop / float(out_packets_sum)) * 100, 2)
+ (packet_drop / float(out_packets_sum)) * 100,
+ self.DROP_PERCENT_ROUND)
except ZeroDivisionError:
LOG.info('No traffic is flowing')
samples['DropPercentage'] = drop_percent
if first_run:
- self.rate = out_packets_sum / duration / num_ifaces
completed = True if drop_percent <= tolerance else False
+ if (first_run and
+ self.rate_unit == tp_base.TrafficProfileConfig.RATE_FPS):
+ self.rate = out_packets_sum / duration / num_ifaces
if drop_percent > tolerance:
self.max_rate = self.rate
'packets_dropped': 2,
}
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if setup_env_helper_type is None:
setup_env_helper_type = AclApproxSetupEnvSetupEnvHelper
-
- super(AclApproxVnf, self).__init__(name, vnfd, setup_env_helper_type, resource_helper_type)
+ super(AclApproxVnf, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
# 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 implementation for generic vnf implementation """
import abc
import six
from yardstick.common import messaging
+from yardstick.common.messaging import consumer
from yardstick.common.messaging import payloads
from yardstick.common.messaging import producer
from yardstick.network_services.helpers.samplevnf_helper import PortPairs
iteration=iteration, kpi=kpi))
+@six.add_metaclass(abc.ABCMeta)
+class GenericVNFEndpoint(consumer.NotificationHandler):
+ """Endpoint class for ``GenericVNFConsumer``"""
+
+ @abc.abstractmethod
+ def runner_method_start_iteration(self, ctxt, **kwargs):
+ """Endpoint when RUNNER_METHOD_START_ITERATION is received
+
+ :param ctxt: (dict) {'id': <Producer ID>}
+ :param kwargs: (dict) ``payloads.RunnerPayload`` context
+ """
+
+ @abc.abstractmethod
+ def runner_method_stop_iteration(self, ctxt, **kwargs):
+ """Endpoint when RUNNER_METHOD_STOP_ITERATION is received
+
+ :param ctxt: (dict) {'id': <Producer ID>}
+ :param kwargs: (dict) ``payloads.RunnerPayload`` context
+ """
+
+
+class GenericVNFConsumer(consumer.MessagingConsumer):
+ """MQ consumer for ``GenericVNF`` derived classes"""
+
+ def __init__(self, ctx_ids, endpoints):
+ if not isinstance(endpoints, list):
+ endpoints = [endpoints]
+ super(GenericVNFConsumer, self).__init__(messaging.TOPIC_RUNNER,
+ ctx_ids, endpoints)
+
+
@six.add_metaclass(abc.ABCMeta)
class GenericVNF(object):
"""Class providing file-like API for generic VNF implementation
UPLINK = PortPairs.UPLINK
DOWNLINK = PortPairs.DOWNLINK
- def __init__(self, name, vnfd):
+ def __init__(self, name, vnfd, task_id):
self.name = name
+ self._task_id = task_id
self.vnfd_helper = VnfdHelper(vnfd)
# List of statistics we can obtain from this VNF
# - ETSI MANO 6.3.1.1 monitoring_parameter
@six.add_metaclass(abc.ABCMeta)
class GenericTrafficGen(GenericVNF):
- """ Class providing file-like API for generic traffic generator """
+ """Class providing file-like API for generic traffic generator"""
- def __init__(self, name, vnfd):
- super(GenericTrafficGen, self).__init__(name, vnfd)
+ def __init__(self, name, vnfd, task_id):
+ super(GenericTrafficGen, self).__init__(name, vnfd, task_id)
self.runs_traffic = True
self.traffic_finished = False
self._mq_producer = None
"packets_dropped": 4,
}
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if setup_env_helper_type is None:
setup_env_helper_type = CgnaptApproxSetupEnvHelper
-
- super(CgnaptApproxVnf, self).__init__(name, vnfd, setup_env_helper_type,
- resource_helper_type)
+ super(CgnaptApproxVnf, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
def _vnf_up_post(self):
super(CgnaptApproxVnf, self)._vnf_up_post()
VNF_PROMPT = "PROX started"
LUA_PARAMETER_NAME = "sut"
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if setup_env_helper_type is None:
setup_env_helper_type = ProxDpdkVnfSetupEnvHelper
self.prev_packets_sent = 0
self.prev_tsc = 0
self.tsc_hz = 0
- super(ProxApproxVnf, self).__init__(name, vnfd, setup_env_helper_type,
- resource_helper_type)
+ super(ProxApproxVnf, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
def _vnf_up_post(self):
self.resource_helper.up_post()
WAIT_TIME = 1
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if setup_env_helper_type is None:
setup_env_helper_type = DpdkVnfSetupEnvHelper
vnfd['mgmt-interface'].pop("pkey", "")
vnfd['mgmt-interface']['password'] = 'password'
- super(RouterVNF, self).__init__(name, vnfd, setup_env_helper_type, resource_helper_type)
+ super(RouterVNF, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
def instantiate(self, scenario_cfg, context_cfg):
self.scenario_helper.scenario_cfg = scenario_cfg
test_timeout = self.options.get('timeout', constants.DEFAULT_VNF_TIMEOUT)
return test_duration if test_duration > test_timeout else test_timeout
+
class SampleVNF(GenericVNF):
""" Class providing file-like API for generic VNF implementation """
APP_NAME = "SampleVNF"
# we run the VNF interactively, so the ssh command will timeout after this long
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
- super(SampleVNF, self).__init__(name, vnfd)
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
+ super(SampleVNF, self).__init__(name, vnfd, task_id)
self.bin_path = get_nsb_option('bin_path', '')
self.scenario_helper = ScenarioHelper(self.name)
APP_NAME = 'Sample'
RUN_WAIT = 1
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
- super(SampleVNFTrafficGen, self).__init__(name, vnfd)
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
+ super(SampleVNFTrafficGen, self).__init__(name, vnfd, task_id)
self.bin_path = get_nsb_option('bin_path', '')
self.scenario_helper = ScenarioHelper(self.name)
class IxLoadTrafficGen(SampleVNFTrafficGen):
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if resource_helper_type is None:
resource_helper_type = IxLoadResourceHelper
- super(IxLoadTrafficGen, self).__init__(name, vnfd, setup_env_helper_type,
- resource_helper_type)
+ super(IxLoadTrafficGen, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
self._result = {}
def run_traffic(self, traffic_profile):
APP_NAME = 'Ping'
RUN_WAIT = 4
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if setup_env_helper_type is None:
setup_env_helper_type = PingSetupEnvHelper
if resource_helper_type is None:
resource_helper_type = PingResourceHelper
-
- super(PingTrafficGen, self).__init__(name, vnfd, setup_env_helper_type,
- resource_helper_type)
+ super(PingTrafficGen, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
self._result = {}
def _check_status(self):
LUA_PARAMETER_NAME = "gen"
WAIT_TIME = 1
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
# don't call superclass, use custom wrapper of ProxApproxVnf
- self._vnf_wrapper = ProxApproxVnf(name, vnfd, setup_env_helper_type, resource_helper_type)
+ self._vnf_wrapper = ProxApproxVnf(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
self.bin_path = get_nsb_option('bin_path', '')
self.name = self._vnf_wrapper.name
self.ssh_helper = self._vnf_wrapper.ssh_helper
APP_NAME = 'Ixia'
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if resource_helper_type is None:
resource_helper_type = IxiaResourceHelper
-
- super(IxiaTrafficGen, self).__init__(name, vnfd, setup_env_helper_type,
- resource_helper_type)
+ super(IxiaTrafficGen, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
self._ixia_traffic_gen = None
self.ixia_file_name = ''
self.vnf_port_pairs = []
traffic for rfc2544 testcase.
"""
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if resource_helper_type is None:
resource_helper_type = TrexRfcResourceHelper
-
- super(TrexTrafficGenRFC, self).__init__(name, vnfd, setup_env_helper_type,
- resource_helper_type)
+ super(TrexTrafficGenRFC, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
APP_NAME = 'TRex'
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if resource_helper_type is None:
resource_helper_type = TrexResourceHelper
-
if setup_env_helper_type is None:
setup_env_helper_type = TrexDpdkVnfSetupEnvHelper
-
- super(TrexTrafficGen, self).__init__(name, vnfd, setup_env_helper_type,
- resource_helper_type)
+ super(TrexTrafficGen, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
def _check_status(self):
return self.resource_helper.check_status()
PIPELINE_COMMAND = REPLAY_PIPELINE_COMMAND
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if resource_helper_type is None:
resource_helper_type = UdpReplayResourceHelper
-
if setup_env_helper_type is None:
setup_env_helper_type = UdpReplaySetupEnvHelper
-
- super(UdpReplayApproxVnf, self).__init__(name, vnfd, setup_env_helper_type,
- resource_helper_type)
+ super(UdpReplayApproxVnf, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
def _build_pipeline_kwargs(self):
ports = self.vnfd_helper.port_pairs.all_ports
'packets_dropped': 3,
}
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if setup_env_helper_type is None:
setup_env_helper_type = FWApproxSetupEnvHelper
-
- super(FWApproxVnf, self).__init__(name, vnfd, setup_env_helper_type, resource_helper_type)
+ super(FWApproxVnf, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
COLLECT_KPI = VPE_COLLECT_KPI
WAIT_TIME = 20
- def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+ def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ resource_helper_type=None):
if setup_env_helper_type is None:
setup_env_helper_type = VpeApproxSetupEnvHelper
-
- super(VpeApproxVnf, self).__init__(name, vnfd, setup_env_helper_type, resource_helper_type)
+ super(VpeApproxVnf, self).__init__(
+ name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
def get_stats(self, *args, **kwargs):
raise NotImplementedError
##############################################################################
import copy
+import re
from oslo_serialization import jsonutils
+import six
from yardstick.common import constants
from yardstick.common import exceptions
SSH_MOUNT_PATH = '/tmp/.ssh/'
IMAGE_DEFAULT = 'openretriever/yardstick'
- COMMAND_DEFAULT = '/bin/bash'
+ COMMAND_DEFAULT = ['/bin/bash', '-c']
RESOURCES = ('requests', 'limits')
PORT_OPTIONS = ('containerPort', 'hostIP', 'hostPort', 'name', 'protocol')
+ IMAGE_PULL_POLICY = ('Always', 'IfNotPresent', 'Never')
def __init__(self, name, ssh_key, **kwargs):
self._name = name
self._ssh_key = ssh_key
self._image = kwargs.get('image', self.IMAGE_DEFAULT)
- self._command = [kwargs.get('command', self.COMMAND_DEFAULT)]
- self._args = kwargs.get('args', [])
+ self._command = self._parse_commands(
+ kwargs.get('command', self.COMMAND_DEFAULT))
+ self._args = self._parse_commands(kwargs.get('args', []))
self._volume_mounts = kwargs.get('volumeMounts', [])
self._security_context = kwargs.get('securityContext')
self._env = kwargs.get('env', [])
self._resources = kwargs.get('resources', {})
self._ports = kwargs.get('ports', [])
+ self._image_pull_policy = kwargs.get('imagePullPolicy')
+ self._tty = kwargs.get('tty')
+ self._stdin = kwargs.get('stdin')
+
+ @staticmethod
+ def _parse_commands(command):
+ if isinstance(command, six.string_types):
+ return [command]
+ elif isinstance(command, list):
+ return command
+ raise exceptions.KubernetesContainerCommandType()
def _create_volume_mounts(self):
"""Return all "volumeMounts" items per container"""
for res in (res for res in self._resources if
res in self.RESOURCES):
container['resources'][res] = self._resources[res]
+ if self._image_pull_policy:
+ if self._image_pull_policy not in self.IMAGE_PULL_POLICY:
+ raise exceptions.KubernetesContainerWrongImagePullPolicy()
+ container['imagePullPolicy'] = self._image_pull_policy
+ if self._stdin is not None:
+ container['stdin'] = self._stdin
+ if self._tty is not None:
+ container['tty'] = self._tty
return container
class ServiceNodePortObject(object):
+ MANDATORY_PARAMETERS = {'port', 'name'}
+ NAME_REGEX = re.compile(r'^[a-z0-9]([-a-z0-9]*[a-z0-9])?$')
+
def __init__(self, name, **kwargs):
"""Service kind "NodePort" object
}
}
- self._add_port(22, protocol='TCP')
+ self._add_port(22, 'ssh', protocol='TCP')
node_ports = copy.deepcopy(kwargs.get('node_ports', []))
for port in node_ports:
+ if not self.MANDATORY_PARAMETERS.issubset(port.keys()):
+ missing_parameters = ', '.join(
+ str(param) for param in
+ (self.MANDATORY_PARAMETERS - set(port.keys())))
+ raise exceptions.KubernetesServiceObjectDefinitionError(
+ missing_parameters=missing_parameters)
port_number = port.pop('port')
- self._add_port(port_number, **port)
+ name = port.pop('name')
+ if not self.NAME_REGEX.match(name):
+ raise exceptions.KubernetesServiceObjectNameError(name=name)
+ self._add_port(port_number, name, **port)
- def _add_port(self, port, protocol=None, name=None, targetPort=None,
+ def _add_port(self, port, name, protocol=None, targetPort=None,
nodePort=None):
- _port = {'port': port}
+ _port = {'port': port,
+ 'name': name}
if protocol:
_port['protocol'] = protocol
- if name:
- _port['name'] = name
if targetPort:
_port['targetPort'] = targetPort
if nodePort:
k8s_utils.create_service(self.template)
def delete(self):
- k8s_utils.delete_service(self._name)
+ k8s_utils.delete_service(self._name, skip_codes=[404])
class CustomResourceDefinitionObject(object):
k8s_utils.create_custom_resource_definition(self._template)
def delete(self):
- k8s_utils.delete_custom_resource_definition(self._name)
+ k8s_utils.delete_custom_resource_definition(self._name, skip_codes=[404])
class NetworkObject(object):
--- /dev/null
+# 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 multiprocessing
+import time
+import uuid
+
+import mock
+
+from yardstick.common import messaging
+from yardstick.common.messaging import payloads
+from yardstick.common.messaging import producer
+from yardstick.network_services.vnf_generic.vnf import base as vnf_base
+from yardstick.tests.functional import base as ft_base
+
+
+class _TrafficGenMQConsumer(vnf_base.GenericTrafficGen,
+ vnf_base.GenericVNFEndpoint):
+
+ def __init__(self, name, vnfd, task_id):
+ vnf_base.GenericTrafficGen.__init__(self, name, vnfd, task_id)
+ self.queue = multiprocessing.Queue()
+ self._id = uuid.uuid1().int
+ vnf_base.GenericVNFEndpoint.__init__(self, self._id, [task_id],
+ self.queue)
+ self._consumer = vnf_base.GenericVNFConsumer([task_id], self)
+ self._consumer.start_rpc_server()
+
+ def run_traffic(self, *args):
+ pass
+
+ def terminate(self):
+ pass
+
+ def collect_kpi(self):
+ pass
+
+ def instantiate(self, *args):
+ pass
+
+ def scale(self, flavor=''):
+ pass
+
+ def runner_method_start_iteration(self, ctxt, **kwargs):
+ if ctxt['id'] in self._ctx_ids:
+ self._queue.put(
+ {'action': messaging.RUNNER_METHOD_START_ITERATION,
+ 'payload': payloads.RunnerPayload.dict_to_obj(kwargs)})
+
+ def runner_method_stop_iteration(self, ctxt, **kwargs):
+ if ctxt['id'] in self._ctx_ids:
+ self._queue.put(
+ {'action': messaging.RUNNER_METHOD_STOP_ITERATION,
+ 'payload': payloads.RunnerPayload.dict_to_obj(kwargs)})
+
+
+class _DummyProducer(producer.MessagingProducer):
+ pass
+
+
+class GenericVNFMQConsumerTestCase(ft_base.BaseFunctionalTestCase):
+
+ def test_fistro(self):
+ vnfd = {'benchmark': {'kpi': mock.ANY},
+ 'vdu': [{'external-interface': 'ext_int'}]
+ }
+ task_id = uuid.uuid1().int
+ tg_obj = _TrafficGenMQConsumer('name_tg', vnfd, task_id)
+ producer = _DummyProducer(messaging.TOPIC_RUNNER, task_id)
+
+ num_messages = 10
+ for i in range(num_messages):
+ pload = payloads.RunnerPayload(version=10, data=i)
+ for method in (messaging.RUNNER_METHOD_START_ITERATION,
+ messaging.RUNNER_METHOD_STOP_ITERATION):
+ producer.send_message(method, pload)
+
+ time.sleep(0.5) # Let consumers attend the calls
+ output = []
+ while not tg_obj.queue.empty():
+ data = tg_obj.queue.get(True, 1)
+ data_dict = {'action': data['action'],
+ 'payload': data['payload'].obj_to_dict()}
+ output.append(data_dict)
+
+ self.assertEqual(num_messages * 2, len(output))
+ for i in range(num_messages):
+ pload = payloads.RunnerPayload(version=10, data=i).obj_to_dict()
+ for method in (messaging.RUNNER_METHOD_START_ITERATION,
+ messaging.RUNNER_METHOD_STOP_ITERATION):
+ reg = {'action': method, 'payload': pload}
+ self.assertIn(reg, output)
def __init__(self):
self.node_port = 30000
self.port = constants.SSH_PORT
+ self.name = 'port_name'
+ self.protocol = 'TCP'
+ self.target_port = constants.SSH_PORT
class Service(object):
def test_get_server(self, mock_get_node_ip, mock_get_pod_by_name):
mock_get_pod_by_name.return_value = Pod()
mock_get_node_ip.return_value = '172.16.10.131'
- with mock.patch.object(self.k8s_context, '_get_service_ports',
- return_value=[NodePort()]):
+ with mock.patch.object(self.k8s_context, '_get_service_ports') as \
+ mock_get_sports:
+ mock_get_sports.return_value = [
+ {'port': constants.SSH_PORT, 'node_port': 30000}]
server = self.k8s_context._get_server('server_name')
self.assertEqual('server_name', server['name'])
self.assertEqual(30000, server['ssh_port'])
name = 'rc_name'
service_ports = self.k8s_context._get_service_ports(name)
mock_get_service_by_name.assert_called_once_with(name + '-service')
- self.assertEqual(30000, service_ports[0].node_port)
+ expected = {'node_port': 30000,
+ 'port': constants.SSH_PORT,
+ 'name': 'port_name',
+ 'protocol': 'TCP',
+ 'target_port': constants.SSH_PORT}
+ self.assertEqual(expected, service_ports[0])
@mock.patch.object(k8s_utils, 'get_service_by_name',
return_value=None)
self._get_file_abspath("tg_trex_tpl.yaml")
self.context_cfg["nodes"]['vnf__1']['VNF model'] = \
self._get_file_abspath("tg_trex_tpl.yaml")
+ self.context_cfg['task_id'] = 'fake_task_id'
vnf = mock.Mock(autospec=GenericVNF)
self.s.get_vnf_impl = mock.Mock(return_value=vnf)
mock_delete_crd.delete_custom_resource_definition.\
assert_called_once_with('name', 'del_obj')
+ @mock.patch.object(client, 'V1DeleteOptions', return_value='del_obj')
+ @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
+ @mock.patch.object(kubernetes_utils, 'LOG')
+ def test_execute_skip_exception(self, mock_log, mock_get_api, mock_delobj):
+ mock_delete_crd = mock.Mock()
+ mock_delete_crd.delete_custom_resource_definition.side_effect = rest.ApiException(
+ status=404)
+
+ mock_get_api.return_value = mock_delete_crd
+ kubernetes_utils.delete_custom_resource_definition('name', skip_codes=[404])
+
+ mock_delobj.assert_called_once()
+ mock_delete_crd.delete_custom_resource_definition.assert_called_once_with(
+ 'name', 'del_obj')
+
+ mock_log.info.assert_called_once()
+
class GetCustomResourceDefinitionTestCase(base.BaseUnitTestCase):
kubernetes_utils.delete_network(
constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,
mock.ANY)
+
+
+class DeletePodTestCase(base.BaseUnitTestCase):
+ @mock.patch.object(kubernetes_utils, 'get_core_api')
+ def test_execute_correct(self, mock_get_api):
+ mock_api = mock.Mock()
+ mock_get_api.return_value = mock_api
+
+ kubernetes_utils.delete_pod("name", body=None)
+ mock_api.delete_namespaced_pod.assert_called_once_with(
+ "name", 'default', None)
+
+ @mock.patch.object(kubernetes_utils, 'get_core_api')
+ def test_execute_exception(self, mock_get_api):
+ mock_api = mock.Mock()
+ mock_api.delete_namespaced_pod.side_effect = rest.ApiException(status=200)
+
+ mock_get_api.return_value = mock_api
+ with self.assertRaises(exceptions.KubernetesApiException):
+ kubernetes_utils.delete_pod(mock.ANY, skip_codes=[404])
+
+ @mock.patch.object(kubernetes_utils, 'get_core_api')
+ def test_execute_skip_exception(self, mock_get_api):
+ mock_api = mock.Mock()
+ mock_api.delete_namespaced_pod.side_effect = rest.ApiException(status=404)
+
+ mock_get_api.return_value = mock_api
+ kubernetes_utils.delete_pod(mock.ANY, skip_codes=[404])
+
+
+class DeleteServiceTestCase(base.BaseUnitTestCase):
+ @mock.patch.object(client, "V1DeleteOptions")
+ @mock.patch.object(kubernetes_utils, 'get_core_api')
+ def test_execute_correct(self, mock_get_api, mock_options):
+ mock_api = mock.Mock()
+ mock_get_api.return_value = mock_api
+ mock_options.return_value = None
+ kubernetes_utils.delete_service("name", "default", None)
+ mock_api.delete_namespaced_service.assert_called_once_with(
+ "name", 'default', None)
+
+ @mock.patch.object(kubernetes_utils, 'get_core_api')
+ def test_execute_exception(self, mock_get_api):
+ mock_api = mock.Mock()
+ mock_api.delete_namespaced_service.side_effect = rest.ApiException(status=200)
+
+ mock_get_api.return_value = mock_api
+ with self.assertRaises(exceptions.KubernetesApiException):
+ kubernetes_utils.delete_service(mock.ANY, skip_codes=[404])
+
+ @mock.patch.object(kubernetes_utils, 'get_core_api')
+ @mock.patch.object(kubernetes_utils, 'LOG')
+ def test_execute_skip_exception(self, mock_log, mock_get_api):
+ mock_api = mock.Mock()
+ mock_api.delete_namespaced_service.side_effect = rest.ApiException(status=404)
+
+ mock_get_api.return_value = mock_api
+ kubernetes_utils.delete_service(mock.ANY, skip_codes=[404])
+ mock_log.info.assert_called_once()
'id': 1,
'bidir': 'False',
'duration': 60,
- 'iload': '100',
+ 'rate': 10000.5,
+ 'rate_unit': 'fps',
'outer_l2': {
'framesize': {'64B': '25', '256B': '75'}
},
'id': 2,
'bidir': 'False',
'duration': 60,
- 'iload': '100',
+ 'rate': 75.2,
+ 'rate_unit': '%',
'outer_l2': {
'framesize': {'128B': '35', '1024B': '65'}
},
self.assertEqual(6, len(ixnet_gen.ixnet.setMultiAttribute.mock_calls))
self.assertEqual(4, len(mock_update_frame.mock_calls))
+ ixnet_gen.ixnet.setMultiAttribute.assert_has_calls(
+ [mock.call('cfg_element/frameRate', '-rate', 10000.5,
+ '-type', 'framesPerSecond'),
+ mock.call('cfg_element/frameRate', '-rate', 75.2, '-type',
+ 'percentLineRate')],
+ any_order=True)
def test_update_frame_flow_not_present(self):
ixnet_gen = ixnet_api.IxNextgen()
self.assertEqual({'64B': 100}, tp_config_obj.packet_sizes)
self.assertEqual(base.TrafficProfileConfig.DEFAULT_SCHEMA,
tp_config_obj.schema)
- self.assertEqual(base.TrafficProfileConfig.DEFAULT_FRAME_RATE,
+ self.assertEqual(float(base.TrafficProfileConfig.DEFAULT_FRAME_RATE),
tp_config_obj.frame_rate)
self.assertEqual(base.TrafficProfileConfig.DEFAULT_DURATION,
tp_config_obj.duration)
+
+ def test__parse_rate(self):
+ tp_config = {'traffic_profile': {'packet_sizes': {'64B': 100}}}
+ tp_config_obj = base.TrafficProfileConfig(tp_config)
+ self.assertEqual((100.0, 'fps'), tp_config_obj._parse_rate('100 '))
+ self.assertEqual((200.5, 'fps'), tp_config_obj._parse_rate('200.5'))
+ self.assertEqual((300.8, 'fps'), tp_config_obj._parse_rate('300.8fps'))
+ self.assertEqual((400.2, 'fps'),
+ tp_config_obj._parse_rate('400.2 fps'))
+ self.assertEqual((500.3, '%'), tp_config_obj._parse_rate('500.3%'))
+ self.assertEqual((600.1, '%'), tp_config_obj._parse_rate('600.1 %'))
+
+ def test__parse_rate_exception(self):
+ tp_config = {'traffic_profile': {'packet_sizes': {'64B': 100}}}
+ tp_config_obj = base.TrafficProfileConfig(tp_config)
+ with self.assertRaises(exceptions.TrafficProfileRate):
+ tp_config_obj._parse_rate('100Fps')
+ with self.assertRaises(exceptions.TrafficProfileRate):
+ tp_config_obj._parse_rate('100 kbps')
self.assertTrue(completed)
self.assertEqual(23.0, samples['TxThroughput'])
self.assertEqual(21.0, samples['RxThroughput'])
- self.assertEqual(0.1, samples['DropPercentage'])
+ self.assertEqual(0.099651, samples['DropPercentage'])
def test_get_drop_percentage_over_drop_percentage(self):
samples = {'iface_name_1':
self.assertFalse(completed)
self.assertEqual(23.0, samples['TxThroughput'])
self.assertEqual(21.0, samples['RxThroughput'])
- self.assertEqual(0.1, samples['DropPercentage'])
+ self.assertEqual(0.099651, samples['DropPercentage'])
self.assertEqual(rfc2544_profile.rate, rfc2544_profile.max_rate)
def test_get_drop_percentage_under_drop_percentage(self):
self.assertFalse(completed)
self.assertEqual(23.0, samples['TxThroughput'])
self.assertEqual(21.0, samples['RxThroughput'])
- self.assertEqual(0.1, samples['DropPercentage'])
+ self.assertEqual(0.099651, samples['DropPercentage'])
self.assertEqual(rfc2544_profile.rate, rfc2544_profile.min_rate)
@mock.patch.object(ixia_rfc2544.LOG, 'info')
self.assertTrue(completed)
self.assertEqual(23.0, samples['TxThroughput'])
self.assertEqual(21.0, samples['RxThroughput'])
- self.assertEqual(0.1, samples['DropPercentage'])
+ self.assertEqual(0.099651, samples['DropPercentage'])
self.assertEqual(33.45, rfc2544_profile.rate)
# 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 unittest
import mock
import re
import copy
-from yardstick.tests import STL_MOCKS
-from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
from yardstick.common import utils
from yardstick.common import exceptions
from yardstick.benchmark.contexts import base as ctx_base
-
-
-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 acl_vnf
- from yardstick.network_services.vnf_generic.vnf.base import VnfdHelper
- from yardstick.network_services.nfvi.resource import ResourceProfile
- from yardstick.network_services.vnf_generic.vnf.acl_vnf import AclApproxSetupEnvSetupEnvHelper
+from yardstick.network_services.vnf_generic.vnf import acl_vnf
+from yardstick.network_services.vnf_generic.vnf.base import VnfdHelper
+from yardstick.network_services.nfvi.resource import ResourceProfile
+from yardstick.network_services.vnf_generic.vnf.acl_vnf import AclApproxSetupEnvSetupEnvHelper
+from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
TEST_FILE_YAML = 'nsb_test_case.yaml'
def test___init__(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd)
+ acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd, 'task_id')
self.assertIsNone(acl_approx_vnf._vnf_process)
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time")
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd)
+ acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd, 'task_id')
acl_approx_vnf.scenario_helper.scenario_cfg = {
'nodes': {acl_approx_vnf.name: "mock"}
}
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd)
+ acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd, 'task_id')
acl_approx_vnf.q_in = mock.MagicMock()
acl_approx_vnf.q_out = mock.MagicMock()
acl_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd)
+ acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd, 'task_id')
acl_approx_vnf.q_in = mock.MagicMock()
acl_approx_vnf.q_out = mock.MagicMock()
acl_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd)
+ acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd, 'task_id')
acl_approx_vnf._build_config = mock.MagicMock()
acl_approx_vnf.queue_wrapper = mock.MagicMock()
acl_approx_vnf.scenario_helper.scenario_cfg = self.scenario_cfg
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd)
+ acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd, 'task_id')
acl_approx_vnf.deploy_helper = mock.MagicMock()
acl_approx_vnf.resource_helper = mock.MagicMock()
acl_approx_vnf._build_config = mock.MagicMock()
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd)
+ acl_approx_vnf = acl_vnf.AclApproxVnf(name, vnfd, 'task_id')
acl_approx_vnf._vnf_process = mock.MagicMock()
acl_approx_vnf._vnf_process.terminate = mock.Mock()
acl_approx_vnf.used_drivers = {"01:01.0": "i40e",
# 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 multiprocessing
import os
from yardstick.common.messaging import payloads
from yardstick.network_services.vnf_generic.vnf import base
from yardstick.ssh import SSH
+from yardstick.tests.unit import base as ut_base
IP_PIPELINE_CFG_FILE_TPL = ("arp_route_tbl = ({port0_local_ip_hex},"
self.assertIsNotNone(queue_file_wrapper.q_out.empty())
-class TestGenericVNF(unittest.TestCase):
+class TestGenericVNF(ut_base.BaseUnitTestCase):
def test_definition(self):
"""Make sure that the abstract class cannot be instantiated"""
with self.assertRaises(TypeError) as exc:
# pylint: disable=abstract-class-instantiated
- base.GenericVNF('vnf1', VNFD['vnfd:vnfd-catalog']['vnfd'][0])
+ base.GenericVNF('vnf1', VNFD['vnfd:vnfd-catalog']['vnfd'][0],
+ 'task_id')
- msg = ("Can't instantiate abstract class GenericVNF with abstract methods "
- "collect_kpi, instantiate, scale, start_collect, "
+ msg = ("Can't instantiate abstract class GenericVNF with abstract "
+ "methods collect_kpi, instantiate, scale, start_collect, "
"stop_collect, terminate, wait_for_instantiate")
self.assertEqual(msg, str(exc.exception))
-class GenericTrafficGenTestCase(unittest.TestCase):
+class GenericTrafficGenTestCase(ut_base.BaseUnitTestCase):
def test_definition(self):
"""Make sure that the abstract class cannot be instantiated"""
name = 'vnf1'
with self.assertRaises(TypeError) as exc:
# pylint: disable=abstract-class-instantiated
- base.GenericTrafficGen(name, vnfd)
+ base.GenericTrafficGen(name, vnfd, 'task_id')
msg = ("Can't instantiate abstract class GenericTrafficGen with "
"abstract methods collect_kpi, instantiate, run_traffic, "
"scale, terminate")
vnfd = {'benchmark': {'kpi': mock.ANY},
'vdu': [{'external-interface': 'ext_int'}]
}
- tg = _DummyGenericTrafficGen('name', vnfd)
+ tg = _DummyGenericTrafficGen('name', vnfd, 'task_id')
tg._mq_producer = mock.Mock()
tg._mq_producer.id = 'fake_id'
self.assertEqual('fake_id', tg.get_mq_producer_id())
-class TrafficGeneratorProducerTestCase(unittest.TestCase):
+class TrafficGeneratorProducerTestCase(ut_base.BaseUnitTestCase):
@mock.patch.object(oslo_messaging, 'Target', return_value='rpc_target')
@mock.patch.object(oslo_messaging, 'RPCClient')
'tg_pload')
mock_tg_payload.assert_called_once_with(version=30, iteration=100,
kpi={'k': 'v'})
+
+
+class GenericVNFConsumerTestCase(ut_base.BaseUnitTestCase):
+
+ def test__init(self):
+ endpoints = 'endpoint_1'
+ _ids = [uuid.uuid1().int]
+ gvnf_consumer = base.GenericVNFConsumer(_ids, endpoints)
+ self.assertEqual(_ids, gvnf_consumer._ids)
+ self.assertEqual([endpoints], gvnf_consumer._endpoints)
def test___init__(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd)
+ cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd, 'task_id')
self.assertIsNone(cgnapt_approx_vnf._vnf_process)
@mock.patch.object(process, 'check_if_process_failed')
@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]
- cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd)
+ cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd, 'task_id')
cgnapt_approx_vnf.scenario_helper.scenario_cfg = {
'nodes': {cgnapt_approx_vnf.name: "mock"}
}
@mock.patch.object(time, 'sleep')
def test_vnf_execute_command(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd)
+ cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd, 'task_id')
cgnapt_approx_vnf.q_in = mock.Mock()
cgnapt_approx_vnf.q_out = mock.Mock()
cgnapt_approx_vnf.q_out.qsize = mock.Mock(return_value=0)
def test_get_stats(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd)
+ cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd, 'task_id')
with mock.patch.object(cgnapt_approx_vnf, 'vnf_execute') as mock_exec:
mock_exec.return_value = 'output'
self.assertEqual('output', cgnapt_approx_vnf.get_stats())
def test_run_vcgnapt(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd)
+ cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd, 'task_id')
cgnapt_approx_vnf.ssh_helper = mock.Mock()
cgnapt_approx_vnf.setup_helper = mock.Mock()
with mock.patch.object(cgnapt_approx_vnf, '_build_config'), \
@mock.patch.object(ctx_base.Context, 'get_context_from_server')
def test_instantiate(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd)
+ cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd, 'task_id')
cgnapt_approx_vnf.deploy_helper = mock.MagicMock()
cgnapt_approx_vnf.resource_helper = mock.MagicMock()
cgnapt_approx_vnf._build_config = mock.MagicMock()
def test__vnf_up_post(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
self.scenario_cfg['options'][name]['napt'] = 'static'
- cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd)
+ cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd, 'task_id')
cgnapt_approx_vnf.vnf_execute = mock.Mock()
cgnapt_approx_vnf.scenario_helper.scenario_cfg = self.scenario_cfg
with mock.patch.object(cgnapt_approx_vnf, 'setup_helper') as \
def test__vnf_up_post_short(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd)
+ cgnapt_approx_vnf = cgnapt_vnf.CgnaptApproxVnf(name, vnfd, 'task_id')
cgnapt_approx_vnf.scenario_helper.scenario_cfg = self.scenario_cfg
cgnapt_approx_vnf._vnf_up_post()
@mock.patch(SSH_HELPER)
def test___init__(self, ssh, *args):
mock_ssh(ssh)
- prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0)
+ prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0, 'task_id')
self.assertIsNone(prox_approx_vnf._vnf_process)
@mock.patch.object(ctx_base.Context, 'get_physical_node_from_server', return_value='mock_node')
def test_collect_kpi_no_client(self, ssh, *args):
mock_ssh(ssh)
- prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0)
+ prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0, 'task_id')
prox_approx_vnf.scenario_helper.scenario_cfg = {
'nodes': {prox_approx_vnf.name: "mock"}
}
[2, 1, 2, 3, 4, 5], [3, 1, 2, 3, 4, 5]]
resource_helper.collect_collectd_kpi.return_value = {'core': {'result': 234}}
- prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0)
+ prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0, 'task_id')
prox_approx_vnf.scenario_helper.scenario_cfg = {
'nodes': {prox_approx_vnf.name: "mock"}
}
mock_ssh(ssh)
resource_helper = mock.MagicMock()
- prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, deepcopy(self.VNFD0))
+ prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, deepcopy(self.VNFD0),
+ 'task_id')
prox_approx_vnf.scenario_helper.scenario_cfg = {
'nodes': {prox_approx_vnf.name: "mock"}
}
def test_run_prox(self, ssh, *_):
mock_ssh(ssh)
- prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0)
+ prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0, 'task_id')
prox_approx_vnf.scenario_helper.scenario_cfg = self.SCENARIO_CFG
prox_approx_vnf.ssh_helper.join_bin_path.return_value = '/tool_path12/tool_file34'
prox_approx_vnf.setup_helper.remote_path = 'configs/file56.cfg'
@mock.patch(SSH_HELPER)
def bad_test_instantiate(self, *args):
- prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0)
+ prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0, 'task_id')
prox_approx_vnf.scenario_helper = mock.MagicMock()
prox_approx_vnf.setup_helper = mock.MagicMock()
# we can't mock super
@mock.patch(SSH_HELPER)
def test_wait_for_instantiate_panic(self, ssh, *args):
mock_ssh(ssh, exec_result=(1, "", ""))
- prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0)
+ prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0, 'task_id')
prox_approx_vnf._vnf_process = mock.MagicMock(**{"is_alive.return_value": True})
prox_approx_vnf._run_prox = mock.Mock(return_value=0)
prox_approx_vnf.WAIT_TIME = 0
@mock.patch(SSH_HELPER)
def test_terminate(self, ssh, *args):
mock_ssh(ssh)
- prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0)
+ prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0, 'task_id')
prox_approx_vnf._vnf_process = mock.MagicMock()
prox_approx_vnf._vnf_process.terminate = mock.Mock()
prox_approx_vnf.ssh_helper = mock.MagicMock()
@mock.patch(SSH_HELPER)
def test__vnf_up_post(self, ssh, *args):
mock_ssh(ssh)
- prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0)
+ prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0, 'task_id')
prox_approx_vnf.resource_helper = resource_helper = mock.Mock()
prox_approx_vnf._vnf_up_post()
@mock.patch(SSH_HELPER)
def test_vnf_execute_oserror(self, ssh, *args):
mock_ssh(ssh)
- prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0)
+ prox_approx_vnf = prox_vnf.ProxApproxVnf(NAME, self.VNFD0, 'task_id')
prox_approx_vnf.resource_helper = resource_helper = mock.Mock()
resource_helper.execute.side_effect = OSError(errno.EPIPE, "")
# 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 unittest
import mock
-from yardstick.tests import STL_MOCKS
from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
from yardstick.benchmark.contexts import base as ctx_base
-
-
-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.router_vnf import RouterVNF
+from yardstick.network_services.vnf_generic.vnf.router_vnf import RouterVNF
TEST_FILE_YAML = 'nsb_test_case.yaml'
def test___init__(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- router_vnf = RouterVNF(name, vnfd)
+ router_vnf = RouterVNF(name, vnfd, 'task_id')
self.assertIsNone(router_vnf._vnf_process)
def test_get_stats(self):
m = mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- router_vnf = RouterVNF(name, vnfd)
+ router_vnf = RouterVNF(name, vnfd, 'task_id')
router_vnf.scenario_helper.scenario_cfg = {
'nodes': {router_vnf.name: "mock"}
}
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- router_vnf = RouterVNF(name, vnfd)
+ router_vnf = RouterVNF(name, vnfd, 'task_id')
router_vnf.scenario_helper.scenario_cfg = self.scenario_cfg
router_vnf._run()
router_vnf.ssh_helper.drop_connection.assert_called_once()
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- router_vnf = RouterVNF(name, vnfd)
+ router_vnf = RouterVNF(name, vnfd, 'task_id')
router_vnf.WAIT_TIME = 0
router_vnf.INTERFACE_WAIT = 0
self.scenario_cfg.update({"nodes": {"vnf__1": ""}})
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- router_vnf = RouterVNF(name, vnfd)
+ router_vnf = RouterVNF(name, vnfd, 'task_id')
router_vnf._vnf_process = mock.MagicMock()
router_vnf._vnf_process.terminate = mock.Mock()
self.assertIsNone(router_vnf.terminate())
}
def test___init__(self):
- sample_vnf = SampleVNF('vnf1', self.VNFD_0)
+ sample_vnf = SampleVNF('vnf1', self.VNFD_0, 'task_id')
self.assertEqual(sample_vnf.name, 'vnf1')
self.assertDictEqual(sample_vnf.vnfd_helper, self.VNFD_0)
class MyResourceHelper(ResourceHelper):
pass
- sample_vnf = SampleVNF('vnf1', self.VNFD_0, MySetupEnvHelper, MyResourceHelper)
+ sample_vnf = SampleVNF('vnf1', self.VNFD_0, 'task_id',
+ MySetupEnvHelper, MyResourceHelper)
self.assertEqual(sample_vnf.name, 'vnf1')
self.assertDictEqual(sample_vnf.vnfd_helper, self.VNFD_0)
@mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.Process')
def test__start_vnf(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
sample_vnf._run = mock.Mock()
self.assertIsNone(sample_vnf.queue_wrapper)
}
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
sample_vnf.scenario_helper.scenario_cfg = {
'nodes': {sample_vnf.name: 'mock'}
}
'plugin1': {'param': 1}}}
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf__0', vnfd)
+ sample_vnf = SampleVNF('vnf__0', vnfd, 'task_id')
sample_vnf._update_collectd_options(scenario_cfg, context_cfg)
self.assertEqual(sample_vnf.setup_helper.collectd_options, expected)
'plugin1': {'param': 1}}}
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
sample_vnf._update_options(options2, options1)
self.assertEqual(options2, expected)
]
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
sample_vnf.APP_NAME = 'sample1'
sample_vnf.WAIT_TIME_FOR_SCRIPT = 0
sample_vnf._start_server = mock.Mock(return_value=0)
]
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
sample_vnf.APP_NAME = 'sample1'
sample_vnf.q_out = mock.Mock()
sample_vnf.q_out.qsize.side_effect = iter(queue_size_list)
def test_terminate_without_vnf_process(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
sample_vnf.APP_NAME = 'sample1'
sample_vnf.vnf_execute = mock.Mock()
sample_vnf.ssh_helper = mock.Mock()
def test_get_stats(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
sample_vnf.APP_NAME = 'sample1'
sample_vnf.APP_WORD = 'sample1'
sample_vnf.vnf_execute = mock.Mock(return_value='the stats')
@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 = SampleVNF('vnf1', vnfd, 'task_id')
sample_vnf.scenario_helper.scenario_cfg = {
'nodes': {sample_vnf.name: "mock"}
}
@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 = SampleVNF('vnf1', vnfd, 'task_id')
sample_vnf.scenario_helper.scenario_cfg = {
'nodes': {sample_vnf.name: "mock"}
}
def test_scale(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
self.assertRaises(y_exceptions.FunctionNotImplemented,
sample_vnf.scale)
test_cmd = 'test cmd'
run_kwargs = {'arg1': 'val1', 'arg2': 'val2'}
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf = SampleVNF('vnf1', vnfd, 'task_id')
sample_vnf.ssh_helper = mock.Mock()
sample_vnf.setup_helper = mock.Mock()
with mock.patch.object(sample_vnf, '_build_config',
}
def test__check_status(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
with self.assertRaises(NotImplementedError):
sample_vnf_tg._check_status()
def test_listen_traffic(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
sample_vnf_tg.listen_traffic(mock.Mock())
def test_verify_traffic(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
sample_vnf_tg.verify_traffic(mock.Mock())
def test_terminate(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
sample_vnf_tg._traffic_process = mock.Mock()
sample_vnf_tg._tg_process = mock.Mock()
sample_vnf_tg.terminate()
def test__wait_for_process(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
with mock.patch.object(sample_vnf_tg, '_check_status',
return_value=0) as mock_status, \
mock.patch.object(sample_vnf_tg, '_tg_process') as mock_proc:
mock_status.assert_called_once()
def test__wait_for_process_not_alive(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
with mock.patch.object(sample_vnf_tg, '_tg_process') as mock_proc:
mock_proc.is_alive.return_value = False
self.assertRaises(RuntimeError, sample_vnf_tg._wait_for_process)
mock_proc.is_alive.assert_called_once()
def test__wait_for_process_delayed(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
with mock.patch.object(sample_vnf_tg, '_check_status',
side_effect=[1, 0]) as mock_status, \
mock.patch.object(sample_vnf_tg,
mock_status.assert_has_calls([mock.call(), mock.call()])
def test_scale(self):
- sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0)
+ sample_vnf_tg = SampleVNFTrafficGen('tg1', self.VNFD_0, 'task_id')
self.assertRaises(y_exceptions.FunctionNotImplemented,
sample_vnf_tg.scale)
def test___init__(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd)
+ ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd, 'task_id')
self.assertIsNone(ixload_traffic_gen.resource_helper.data)
@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]
- ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd)
+ ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd, 'task_id')
ixload_traffic_gen.scenario_helper.scenario_cfg = {
'nodes': {ixload_traffic_gen.name: "mock"}
}
def test_listen_traffic(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd)
+ ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd, 'task_id')
self.assertIsNone(ixload_traffic_gen.listen_traffic({}))
@mock.patch.object(utils, 'find_relative_file')
@mock.patch.object(tg_ixload, 'shutil')
def test_instantiate(self, mock_shutil, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd)
+ ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd, 'task_id')
scenario_cfg = {'tc': "nsb_test_case",
'ixia_profile': "ixload.cfg",
'task_path': "/path/to/task"}
vnfd['mgmt-interface'].update({'tg-config': {}})
vnfd['mgmt-interface']['tg-config'].update({'ixchassis': '1.1.1.1'})
vnfd['mgmt-interface']['tg-config'].update({'py_bin_path': '/root'})
- sut = tg_ixload.IxLoadTrafficGen(NAME, vnfd)
+ sut = tg_ixload.IxLoadTrafficGen(NAME, vnfd, 'task_id')
sut.connection = mock.Mock()
sut._traffic_runner = mock.Mock(return_value=0)
result = sut.run_traffic(mock_traffic_profile)
vnfd['mgmt-interface'].update({'tg-config': {}})
vnfd['mgmt-interface']['tg-config'].update({'ixchassis': '1.1.1.1'})
vnfd['mgmt-interface']['tg-config'].update({'py_bin_path': '/root'})
- sut = tg_ixload.IxLoadTrafficGen(NAME, vnfd)
+ sut = tg_ixload.IxLoadTrafficGen(NAME, vnfd, 'task_id')
sut.connection = mock.Mock()
sut._traffic_runner = mock.Mock(return_value=0)
subprocess.call(['touch', '/tmp/1.csv'])
def test_terminate(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd)
+ ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd, 'task_id')
self.assertIsNone(ixload_traffic_gen.terminate())
def test_parse_csv_read(self):
'HTTP Transaction Rate': True,
}
http_reader = [kpi_data]
- ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd)
+ ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd, 'task_id')
result = ixload_traffic_gen.resource_helper.result
ixload_traffic_gen.resource_helper.parse_csv_read(http_reader)
for k_left, k_right in tg_ixload.IxLoadResourceHelper.KPI_LIST.items():
'HTTP Connection Rate': 4,
'HTTP Transaction Rate': 5,
}]
- ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd)
+ ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd, 'task_id')
init_value = ixload_traffic_gen.resource_helper.result
ixload_traffic_gen.resource_helper.parse_csv_read(http_reader)
self.assertDictEqual(ixload_traffic_gen.resource_helper.result,
'HTTP Concurrent Connections': 3,
'HTTP Transaction Rate': 5,
}]
- ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd)
+ ixload_traffic_gen = tg_ixload.IxLoadTrafficGen(NAME, vnfd, 'task_id')
+
with self.assertRaises(KeyError):
ixload_traffic_gen.resource_helper.parse_csv_read(http_reader)
import unittest
from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
-from yardstick.tests import STL_MOCKS
from yardstick.benchmark.contexts import base as ctx_base
+from yardstick.network_services.vnf_generic.vnf.tg_ping import PingParser
+from yardstick.network_services.vnf_generic.vnf.tg_ping import PingTrafficGen
+from yardstick.network_services.vnf_generic.vnf.tg_ping import PingResourceHelper
+from yardstick.network_services.vnf_generic.vnf.tg_ping import PingSetupEnvHelper
+from yardstick.network_services.vnf_generic.vnf.vnf_ssh_helper import VnfSshHelper
-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.tg_ping import PingParser
- from yardstick.network_services.vnf_generic.vnf.tg_ping import PingTrafficGen
- from yardstick.network_services.vnf_generic.vnf.tg_ping import PingResourceHelper
- from yardstick.network_services.vnf_generic.vnf.tg_ping import PingSetupEnvHelper
- from yardstick.network_services.vnf_generic.vnf.vnf_ssh_helper import VnfSshHelper
+SSH_HELPER = "yardstick.network_services.vnf_generic.vnf.sample_vnf.VnfSshHelper"
class TestPingResourceHelper(unittest.TestCase):
@mock.patch("yardstick.ssh.SSH")
def test___init__(self, ssh):
ssh.from_node.return_value.execute.return_value = 0, "success", ""
- ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0)
+ ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0, 'task_id')
self.assertIsInstance(ping_traffic_gen.setup_helper, PingSetupEnvHelper)
self.assertIsInstance(ping_traffic_gen.resource_helper, PingResourceHelper)
(0, 'if_name_2', ''),
]
ssh.from_node.return_value.execute.side_effect = iter(execute_result_data)
- ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0)
+ ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0, 'task_id')
ext_ifs = ping_traffic_gen.vnfd_helper.interfaces
self.assertNotEqual(ext_ifs[0]['virtual-interface']['local_iface_name'], 'if_name_1')
self.assertNotEqual(ext_ifs[1]['virtual-interface']['local_iface_name'], 'if_name_2')
def test_collect_kpi(self, ssh, *args):
mock_ssh(ssh, exec_result=(0, "success", ""))
- ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0)
+ ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0, 'task_id')
ping_traffic_gen.scenario_helper.scenario_cfg = {
'nodes': {ping_traffic_gen.name: "mock"}
}
@mock.patch(SSH_HELPER)
def test_instantiate(self, ssh):
mock_ssh(ssh, spec=VnfSshHelper, exec_result=(0, "success", ""))
- ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0)
+ ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0, 'task_id')
ping_traffic_gen.setup_helper.ssh_helper = mock.MagicMock(
**{"execute.return_value": (0, "xe0_fake", "")})
self.assertIsInstance(ping_traffic_gen.ssh_helper, mock.Mock)
self.assertIsNotNone(ping_traffic_gen._result)
def test_listen_traffic(self):
- ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0)
+ ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0, 'task_id')
self.assertIsNone(ping_traffic_gen.listen_traffic({}))
@mock.patch("yardstick.ssh.SSH")
ssh.from_node.return_value.execute.return_value = 0, "success", ""
ssh.from_node.return_value.run.return_value = 0, "success", ""
- ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0)
+ ping_traffic_gen = PingTrafficGen('vnf1', self.VNFD_0, 'task_id')
self.assertIsNone(ping_traffic_gen.terminate())
import mock
from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
-from yardstick.tests import STL_MOCKS
from yardstick.benchmark.contexts import base as ctx_base
+from yardstick.network_services.vnf_generic.vnf.tg_prox import ProxTrafficGen
+from yardstick.network_services.traffic_profile.base import TrafficProfile
SSH_HELPER = 'yardstick.network_services.vnf_generic.vnf.sample_vnf.VnfSshHelper'
NAME = 'vnf__1'
-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.tg_prox import ProxTrafficGen
- from yardstick.network_services.traffic_profile.base import TrafficProfile
-
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.time')
class TestProxTrafficGen(unittest.TestCase):
@mock.patch(SSH_HELPER)
def test___init__(self, ssh, *args):
mock_ssh(ssh)
- prox_traffic_gen = ProxTrafficGen(NAME, self.VNFD0)
+ prox_traffic_gen = ProxTrafficGen(NAME, self.VNFD0, 'task_id')
self.assertIsNone(prox_traffic_gen._tg_process)
self.assertIsNone(prox_traffic_gen._traffic_process)
@mock.patch(SSH_HELPER)
def test_collect_kpi(self, ssh, *args):
mock_ssh(ssh)
- prox_traffic_gen = ProxTrafficGen(NAME, self.VNFD0)
+ prox_traffic_gen = ProxTrafficGen(NAME, self.VNFD0, 'task_id')
prox_traffic_gen.scenario_helper.scenario_cfg = {
'nodes': {prox_traffic_gen.name: "mock"}
}
mock_traffic_profile.params = self.TRAFFIC_PROFILE
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- prox_traffic_gen = ProxTrafficGen(NAME, vnfd)
+ prox_traffic_gen = ProxTrafficGen(NAME, vnfd, 'task_id')
ssh_helper = mock.MagicMock(
**{"execute.return_value": (0, "", ""), "bin_path": ""})
prox_traffic_gen.ssh_helper = ssh_helper
mock_traffic_profile.params = self.TRAFFIC_PROFILE
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sut = ProxTrafficGen(NAME, vnfd)
+ sut = ProxTrafficGen(NAME, vnfd, 'task_id')
sut._get_socket = mock.MagicMock()
sut.ssh_helper = mock.Mock()
sut.ssh_helper.run = mock.Mock()
sut.setup_helper.prox_config_dict = {}
- sut._connect_client = mock.Mock(autospec=STLClient)
+ sut._connect_client = mock.Mock(autospec=mock.Mock())
sut._connect_client.get_stats = mock.Mock(return_value="0")
sut._setup_mq_producer = mock.Mock(return_value='mq_producer')
sut._traffic_runner(mock_traffic_profile, mock.ANY)
def test_listen_traffic(self, ssh, *args):
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- prox_traffic_gen = ProxTrafficGen(NAME, vnfd)
+ prox_traffic_gen = ProxTrafficGen(NAME, vnfd, 'task_id')
self.assertIsNone(prox_traffic_gen.listen_traffic(mock.Mock()))
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.socket')
def test_terminate(self, ssh, *args):
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- prox_traffic_gen = ProxTrafficGen(NAME, vnfd)
+ prox_traffic_gen = ProxTrafficGen(NAME, vnfd, 'task_id')
prox_traffic_gen._terminated = mock.MagicMock()
prox_traffic_gen._traffic_process = mock.MagicMock()
prox_traffic_gen._traffic_process.terminate = mock.Mock()
ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
# NOTE(ralonsoh): check the object returned.
- tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
+ tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd, 'task_id')
def test_listen_traffic(self, *args):
with mock.patch("yardstick.ssh.SSH") as ssh:
mock.Mock(return_value=(0, "", ""))
ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
+ ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd,
+ 'task_id')
self.assertIsNone(ixnet_traffic_gen.listen_traffic({}))
@mock.patch.object(ctx_base.Context, 'get_context_from_server', return_value='fake_context')
mock.Mock(return_value=(0, "", ""))
ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
+ ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd,
+ 'task_id')
scenario_cfg = {'tc': "nsb_test_case", "topology": "",
'ixia_profile': "ixload.cfg"}
scenario_cfg.update(
ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
+ ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd,
+ 'task_id')
ixnet_traffic_gen.scenario_helper.scenario_cfg = {
'nodes': {ixnet_traffic_gen.name: "mock"}
}
ssh_mock.execute = \
mock.Mock(return_value=(0, "", ""))
ssh.from_node.return_value = ssh_mock
- ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
+ ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd,
+ 'task_id')
ixnet_traffic_gen._terminated = mock.MagicMock()
ixnet_traffic_gen._terminated.value = 0
ixnet_traffic_gen._ixia_traffic_gen = mock.MagicMock()
def test__check_status(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
+ sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd, 'task_id')
sut._check_status()
@mock.patch("yardstick.ssh.SSH")
mock_traffic_profile.get_drop_percentage.return_value = [
'Completed', samples]
- sut = tg_rfc2544_ixia.IxiaTrafficGen(name, vnfd)
+ sut = tg_rfc2544_ixia.IxiaTrafficGen(name, vnfd, 'task_id')
sut.vnf_port_pairs = [[[0], [1]]]
sut.tc_file_name = self._get_file_abspath(TEST_FILE_YAML)
sut.topology = ""
self._mock_ssh_helper.stop()
def test___init__(self):
- trex_traffic_gen = tg_rfc2544_trex.TrexTrafficGenRFC('vnf1', self.VNFD_0)
+ trex_traffic_gen = tg_rfc2544_trex.TrexTrafficGenRFC(
+ 'vnf1', self.VNFD_0, 'task_id')
self.assertIsNotNone(trex_traffic_gen.resource_helper._terminated.value)
@mock.patch.object(ctx_base.Context, 'get_physical_node_from_server', return_value='mock_node')
def test_collect_kpi(self, *args):
- trex_traffic_gen = tg_rfc2544_trex.TrexTrafficGenRFC('vnf1', self.VNFD_0)
+ trex_traffic_gen = tg_rfc2544_trex.TrexTrafficGenRFC(
+ 'vnf1', self.VNFD_0, 'task_id')
trex_traffic_gen.scenario_helper.scenario_cfg = {
'nodes': {trex_traffic_gen.name: "mock"}
}
mock_traffic_profile.get_traffic_definition.return_value = "64"
mock_traffic_profile.params = self.TRAFFIC_PROFILE
- trex_traffic_gen = tg_rfc2544_trex.TrexTrafficGenRFC('vnf1', self.VNFD_0)
+ trex_traffic_gen = tg_rfc2544_trex.TrexTrafficGenRFC(
+ 'vnf1', self.VNFD_0, 'task_id')
trex_traffic_gen._start_server = mock.Mock(return_value=0)
trex_traffic_gen.resource_helper = mock.MagicMock()
trex_traffic_gen.setup_helper.setup_vnf_environment = mock.MagicMock()
mock_traffic_profile.get_traffic_definition.return_value = "64"
mock_traffic_profile.params = self.TRAFFIC_PROFILE
- trex_traffic_gen = tg_rfc2544_trex.TrexTrafficGenRFC('vnf1', self.VNFD_0)
+ trex_traffic_gen = tg_rfc2544_trex.TrexTrafficGenRFC(
+ 'vnf1', self.VNFD_0, 'task_id')
trex_traffic_gen.resource_helper = mock.MagicMock()
trex_traffic_gen.setup_helper.setup_vnf_environment = mock.MagicMock()
scenario_cfg = {
def test___init__(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
self.assertIsInstance(trex_traffic_gen.resource_helper,
tg_trex.TrexResourceHelper)
@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]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
trex_traffic_gen.scenario_helper.scenario_cfg = {
'nodes': {trex_traffic_gen.name: "mock"}
}
def test_listen_traffic(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
self.assertIsNone(trex_traffic_gen.listen_traffic({}))
@mock.patch.object(ctx_base.Context, 'get_context_from_server', return_value='fake_context')
def test_instantiate(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
trex_traffic_gen._start_server = mock.Mock(return_value=0)
trex_traffic_gen._tg_process = mock.MagicMock()
trex_traffic_gen._tg_process.start = mock.Mock()
@mock.patch.object(ctx_base.Context, 'get_context_from_server', return_value='fake_context')
def test_instantiate_error(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
trex_traffic_gen._start_server = mock.Mock(return_value=0)
trex_traffic_gen._tg_process = mock.MagicMock()
trex_traffic_gen._tg_process.start = mock.Mock()
def test__start_server(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
trex_traffic_gen.ssh_helper = mock.MagicMock()
trex_traffic_gen.resource_helper.ssh_helper = mock.MagicMock()
trex_traffic_gen.scenario_helper.scenario_cfg = {}
def test__start_server_multiple_queues(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
trex_traffic_gen.ssh_helper = mock.MagicMock()
trex_traffic_gen.resource_helper.ssh_helper = mock.MagicMock()
trex_traffic_gen.scenario_helper.scenario_cfg = {
mock_traffic_profile.params = self.TRAFFIC_PROFILE
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- self.sut = tg_trex.TrexTrafficGen(NAME, vnfd)
+ self.sut = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
self.sut.ssh_helper = mock.Mock()
self.sut.ssh_helper.run = mock.Mock()
self.sut._connect_client = mock.Mock()
def test__generate_trex_cfg(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
trex_traffic_gen.resource_helper.ssh_helper = mock.MagicMock()
self.assertIsNone(trex_traffic_gen.resource_helper.generate_cfg())
'local_mac': '00:00:00:00:00:01'},
'vnfd-connection-point-ref': 'xe1',
'name': 'xe1'}]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
trex_traffic_gen.resource_helper.ssh_helper = mock.MagicMock()
trex_traffic_gen.resource_helper.generate_cfg()
trex_traffic_gen.resource_helper._build_ports()
mock_traffic_profile.params = self.TRAFFIC_PROFILE
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- self.sut = tg_trex.TrexTrafficGen(NAME, vnfd)
+ self.sut = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
self.sut.ssh_helper = mock.Mock()
self.sut.ssh_helper.run = mock.Mock()
self.sut._traffic_runner = mock.Mock(return_value=0)
def test_terminate(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
trex_traffic_gen.ssh_helper = mock.MagicMock()
trex_traffic_gen.resource_helper.ssh_helper = mock.MagicMock()
self.assertIsNone(trex_traffic_gen.terminate())
def test__connect_client(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd)
+ trex_traffic_gen = tg_trex.TrexTrafficGen(NAME, vnfd, 'task_id')
client = mock.Mock()
client.connect = mock.Mock(return_value=0)
self.assertIsNotNone(trex_traffic_gen.resource_helper._connect(client))
# 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 unittest
import mock
import os
-from yardstick.tests import STL_MOCKS
from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
from yardstick.benchmark.contexts import base as ctx_base
+from yardstick.network_services.vnf_generic.vnf.udp_replay import UdpReplayApproxVnf
+from yardstick.network_services.vnf_generic.vnf.sample_vnf import ScenarioHelper
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.udp_replay import UdpReplayApproxVnf
- from yardstick.network_services.vnf_generic.vnf.sample_vnf import ScenarioHelper
-
-
TEST_FILE_YAML = 'nsb_test_case.yaml'
-
NAME = "vnf__1"
}
def test___init__(self, *args):
- udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0)
+ udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0,
+ 'task_id')
self.assertIsNone(udp_replay_approx_vnf._vnf_process)
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time")
"Port\t\tRx Packet\t\tTx Packet\t\tRx Pkt Drop\t\tTx Pkt Drop \r\n"\
"0\t\t7374156\t\t7374136\t\t\t0\t\t\t0\r\n" \
"1\t\t7374316\t\t7374315\t\t\t0\t\t\t0\r\n\r\nReplay>\r\r\nReplay>"
- udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, vnfd)
+ udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, vnfd, 'task_id')
udp_replay_approx_vnf.scenario_helper.scenario_cfg = {
'nodes': {udp_replay_approx_vnf.name: "mock"}
}
def test_get_stats(self, ssh, *args):
mock_ssh(ssh)
- udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0)
+ udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0,
+ 'task_id')
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)
nfvi_context.attrs = {'nfvi_type': 'baremetal'}
mock_get_ctx.return_value = nfvi_context
- udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0)
+ udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0,
+ 'task_id')
udp_replay_approx_vnf.queue_wrapper = mock.MagicMock()
udp_replay_approx_vnf.nfvi_context = mock_get_ctx
udp_replay_approx_vnf.nfvi_context.attrs = {'nfvi_type': 'baremetal'}
nfvi_context.attrs = {'nfvi_type': "baremetal"}
mock_get_ctx.return_value = nfvi_context
- udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0)
+ udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0,
+ 'task_id')
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")
def test_run_udp_replay(self, ssh, *args):
mock_ssh(ssh)
- udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0)
+ udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0,
+ 'task_id')
udp_replay_approx_vnf._build_config = mock.MagicMock()
udp_replay_approx_vnf.queue_wrapper = mock.MagicMock()
udp_replay_approx_vnf.scenario_helper = mock.MagicMock()
def test_instantiate(self, ssh, *args):
mock_ssh(ssh)
- udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0)
+ udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0,
+ 'task_id')
udp_replay_approx_vnf.q_out.put("Replay>")
udp_replay_approx_vnf.WAIT_TIME = 0
udp_replay_approx_vnf.setup_helper.setup_vnf_environment = mock.Mock()
@mock.patch('yardstick.ssh.SSH')
@mock.patch(SSH_HELPER)
def test_instantiate_panic(self, *args):
- udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0)
+ udp_replay_approx_vnf = UdpReplayApproxVnf(NAME, self.VNFD_0,
+ 'task_id')
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()
# 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 unittest
import mock
import os
-from yardstick.tests import STL_MOCKS
-from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
-
from yardstick.common import utils
from yardstick.benchmark.contexts import base as ctx_base
-
-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.vfw_vnf import FWApproxVnf
- from yardstick.network_services.nfvi.resource import ResourceProfile
- from yardstick.network_services.vnf_generic.vnf.vfw_vnf import FWApproxSetupEnvHelper
+from yardstick.network_services.vnf_generic.vnf.vfw_vnf import FWApproxVnf
+from yardstick.network_services.nfvi.resource import ResourceProfile
+from yardstick.network_services.vnf_generic.vnf.vfw_vnf import FWApproxSetupEnvHelper
+from yardstick.tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
TEST_FILE_YAML = 'nsb_test_case.yaml'
def test___init__(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
+ vfw_approx_vnf = FWApproxVnf(name, vnfd, 'task_id')
self.assertIsNone(vfw_approx_vnf._vnf_process)
STATS = """\
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 = FWApproxVnf(name, vnfd, 'task_id')
vfw_approx_vnf.scenario_helper.scenario_cfg = {
'nodes': {vfw_approx_vnf.name: "mock"}
}
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
+ vfw_approx_vnf = FWApproxVnf(name, vnfd, 'task_id')
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)
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
+ vfw_approx_vnf = FWApproxVnf(name, vnfd, 'task_id')
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)
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
+ vfw_approx_vnf = FWApproxVnf(name, vnfd, 'task_id')
vfw_approx_vnf._build_config = mock.MagicMock()
vfw_approx_vnf.queue_wrapper = mock.MagicMock()
vfw_approx_vnf.ssh_helper = mock.MagicMock()
mock_ssh(ssh)
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- vfw_approx_vnf = FWApproxVnf(name, vnfd)
+ vfw_approx_vnf = FWApproxVnf(name, vnfd, 'task_id')
vfw_approx_vnf.ssh_helper = ssh
vfw_approx_vnf.deploy_helper = mock.MagicMock()
vfw_approx_vnf.resource_helper = mock.MagicMock()
self._mock_time_sleep.stop()
def test___init__(self):
- vpe_approx_vnf = vpe_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.object(ctx_base.Context, 'get_physical_node_from_server',
resource.amqp_collect_nfvi_kpi.return_value = {'foo': 234}
resource.check_if_system_agent_running.return_value = (1, None)
- vpe_approx_vnf = vpe_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"}
}
resource.check_if_system_agent_running.return_value = 0, '1234'
resource.amqp_collect_nfvi_kpi.return_value = {'foo': 234}
- vpe_approx_vnf = vpe_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"}
}
@mock.patch.object(sample_vnf, 'VnfSshHelper')
def test_vnf_execute(self, ssh):
test_base.mock_ssh(ssh)
- vpe_approx_vnf = vpe_vnf.VpeApproxVnf(NAME, self.VNFD_0)
+ 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)
def test_run_vpe(self, ssh):
test_base.mock_ssh(ssh)
- vpe_approx_vnf = vpe_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',
mock_resource = mock.MagicMock()
- vpe_approx_vnf = vpe_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(
mock_resource = mock.MagicMock()
- vpe_approx_vnf = vpe_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(
mock_resource = mock.MagicMock()
- vpe_approx_vnf = vpe_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
mock_resource = mock.MagicMock()
- vpe_approx_vnf = vpe_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
mock_resource = mock.MagicMock()
- vpe_approx_vnf = vpe_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
def test_terminate(self, ssh):
test_base.mock_ssh(ssh)
- vpe_approx_vnf = vpe_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()
'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
args=args)
expected = {'args': args,
- 'command': [kubernetes.ContainerObject.COMMAND_DEFAULT],
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
'name': 'cname-container',
'volumeMounts': container_obj._create_volume_mounts()}
'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
args=args, securityContext={'key': 'value'})
expected = {'args': args,
- 'command': [kubernetes.ContainerObject.COMMAND_DEFAULT],
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
'name': 'cname-container',
'volumeMounts': container_obj._create_volume_mounts(),
args=args, env=[{'name': 'fake_var_name',
'value': 'fake_var_value'}])
expected = {'args': args,
- 'command': [kubernetes.ContainerObject.COMMAND_DEFAULT],
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
'name': 'cname-container',
'volumeMounts': container_obj._create_volume_mounts(),
'invalid_varible': 'fakeinvalid_varible',
'hostIP': 'fake_port_number'}])
expected = {'args': args,
- 'command': [
- kubernetes.ContainerObject.COMMAND_DEFAULT],
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
'name': 'cname-container',
'volumeMounts': container_obj._create_volume_mounts(),
'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
args=args, resources=resources)
expected = {'args': args,
- 'command': [kubernetes.ContainerObject.COMMAND_DEFAULT],
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
'name': 'cname-container',
'volumeMounts': container_obj._create_volume_mounts(),
'limits': {'key2': 'val2'}}}
self.assertEqual(expected, container_obj.get_container_item())
+ def test_get_container_item_image_pull_policy(self):
+ container_obj = kubernetes.ContainerObject(
+ 'cname', ssh_key='fake_sshkey', imagePullPolicy='Always')
+ expected = {'args': [],
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
+ 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
+ 'name': 'cname-container',
+ 'volumeMounts': container_obj._create_volume_mounts(),
+ 'imagePullPolicy':'Always'}
+ self.assertEqual(expected, container_obj.get_container_item())
+
+ def test_get_container_item_with_tty_stdin(self):
+ args = ['arg1', 'arg2']
+ container_obj = kubernetes.ContainerObject(
+ 'cname', 'fake_sshkey', args=args, tty=False, stdin=True)
+ expected = {'args': args,
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
+ 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
+ 'name': 'cname-container',
+ 'volumeMounts': container_obj._create_volume_mounts(),
+ 'tty': False,
+ 'stdin': True}
+ self.assertEqual(expected, container_obj.get_container_item())
+
+ def test__parse_commands_string(self):
+ container_obj = kubernetes.ContainerObject('cname', 'fake_sshkey')
+ self.assertEqual(['fake command'],
+ container_obj._parse_commands('fake command'))
+
+ def test__parse_commands_list(self):
+ container_obj = kubernetes.ContainerObject('cname', 'fake_sshkey')
+ self.assertEqual(['cmd1', 'cmd2'],
+ container_obj._parse_commands(['cmd1', 'cmd2']))
+
+ def test__parse_commands_exception(self):
+ container_obj = kubernetes.ContainerObject('cname', 'fake_sshkey')
+ with self.assertRaises(exceptions.KubernetesContainerCommandType):
+ container_obj._parse_commands({})
+
class CustomResourceDefinitionObjectTestCase(base.BaseUnitTestCase):
def test__init(self):
with mock.patch.object(kubernetes.ServiceNodePortObject, '_add_port') \
as mock_add_port:
- kubernetes.ServiceNodePortObject('fake_name',
- node_ports=[{'port': 80}])
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'port': 80, 'name': 'web'}])
+
+ mock_add_port.assert_has_calls([mock.call(22, 'ssh', protocol='TCP'),
+ mock.call(80, 'web')])
- mock_add_port.assert_has_calls([mock.call(22, protocol='TCP'),
- mock.call(80)])
+ @mock.patch.object(kubernetes.ServiceNodePortObject, '_add_port')
+ def test__init_missing_mandatory_parameters(self, *args):
+ with self.assertRaises(
+ exceptions.KubernetesServiceObjectDefinitionError):
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'port': 80}])
+ with self.assertRaises(
+ exceptions.KubernetesServiceObjectDefinitionError):
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'name': 'web'}])
+
+ @mock.patch.object(kubernetes.ServiceNodePortObject, '_add_port')
+ def test__init_missing_bad_name(self, *args):
+ with self.assertRaises(
+ exceptions.KubernetesServiceObjectNameError):
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'port': 80, 'name': '-web'}])
+ with self.assertRaises(
+ exceptions.KubernetesServiceObjectNameError):
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'port': 80, 'name': 'Web'}])
+ with self.assertRaises(
+ exceptions.KubernetesServiceObjectNameError):
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'port': 80, 'name': 'web-'}])
def test__add_port(self):
nodeport_object = kubernetes.ServiceNodePortObject('fake_name')
- port_ssh = {'port': 22,
- 'protocol': 'TCP',}
+ port_ssh = {'name': 'ssh',
+ 'port': 22,
+ 'protocol': 'TCP'}
port_definition = {'port': 80,
'protocol': 'TCP',
'name': 'web',
'targetPort': 10080,
'nodePort': 30080}
port = copy.deepcopy(port_definition)
- port.pop('port')
- nodeport_object._add_port(80, **port)
+ _port = port.pop('port')
+ name = port.pop('name')
+ nodeport_object._add_port(_port, name, **port)
self.assertEqual([port_ssh, port_definition],
nodeport_object.template['spec']['ports'])
def test_delete(self, mock_delete_service):
nodeport_object = kubernetes.ServiceNodePortObject('fake_name')
nodeport_object.delete()
- mock_delete_service.assert_called_once_with('fake_name-service')
+ mock_delete_service.assert_called_once_with('fake_name-service',
+ skip_codes=[404])
class KubernetesTemplate(base.BaseUnitTestCase):
Code Review / yardstick.git / commitdiff