# The only case where this field can be empty is when measuring a system that does not run
# OpenStack or when OpenStack APIs are not accessible or OpenStack APis use is not
# desirable. In that case the EXT service chain must be used.
+#
+# If openrc is not admin some parameters are mandatory and must be filled with valid values in config file such as :
+# - availability_zone
+# - hypervisor_hostname
+# - vlans
openrc_file:
# Forwarder to use in nfvbenchvm image. Available options: ['vpp', 'testpmd']
# Name of the availability zone to use for the test VMs
# Must be one of the zones listed by 'nova availability-zone-list'
# availability_zone: 'nova'
+# If openrc is not admin set a valid value
availability_zone:
# To force placement on a given hypervisor, set the name here
# (if multiple names are provided, the first will be used)
# Leave empty to let openstack pick the hypervisor
compute_nodes:
+# If openrc is not admin set a valid value for hypervisor hostname
+# Example of value: hypervisor_hostname: "server1"
+hypervisor_hostname:
# Type of service chain to run, possible options are PVP, PVVP and EXT
# PVP - port to VM to port
#
# Generator profiles are listed in the following format:
# `name`: Traffic generator profile name (use a unique name, no space or special character)
- # DFo not change this field
+ # Do not change this field
# `tool`: Traffic generator tool to be used (currently supported is `TRex`).
# Do not change this field
# `ip`: IP address of the traffic generator.
# is not supported). Use the vtep_vlan option to enable vlan tagging for the VxLAN overlay network.
vlan_tagging: true
-# Used only in the case of EXT chain and no openstack to specify the VLAN IDs to use.
+# Used only in the case of EXT chain and no openstack or not admin access to specify the VLAN IDs to use.
# This property is ignored when OpenStakc is used or in the case of l2-loopback.
# If OpenStack is used leave the list empty, VLAN IDs are retrieved from OpenStack networks using Neutron API.
# If networks are shared across all chains (service_chain_shared_net=true), the list should have exactly 2 values
def get_hostname(self):
"""Get the hypervisor host name running this VNF instance."""
- return getattr(self.instance, 'OS-EXT-SRV-ATTR:hypervisor_hostname')
+ if self.manager.is_admin:
+ hypervisor_hostname = getattr(self.instance, 'OS-EXT-SRV-ATTR:hypervisor_hostname')
+ else:
+ hypervisor_hostname = self.manager.config.hypervisor_hostname
+ if not hypervisor_hostname:
+ raise ChainException('Hypervisor hostname parameter is mandatory')
+ return hypervisor_hostname
def get_host_ip(self):
"""Get the IP address of the host where this instance runs.
def get_hypervisor_name(self):
"""Get hypervisor name (az:hostname) for this VNF instance."""
if self.instance:
- az = getattr(self.instance, 'OS-EXT-AZ:availability_zone')
+ if self.manager.is_admin:
+ az = getattr(self.instance, 'OS-EXT-AZ:availability_zone')
+ else:
+ az = self.manager.config.availability_zone
+ if not az:
+ raise ChainException('Availability zone parameter is mandatory')
hostname = self.get_hostname()
if az:
return az + ':' + hostname
if self.openstack:
# openstack only
session = chain_runner.cred.get_session()
+ self.is_admin = chain_runner.cred.is_admin
self.nova_client = Client(2, session=session)
self.neutron_client = neutronclient.Client('2.0', session=session)
self.glance_client = glanceclient.Client('2', session=session)
else:
# Make sure all instances are active before proceeding
self._ensure_instances_active()
+ # network API call do not show VLANS ID if not admin read from config
+ if not self.is_admin:
+ self._get_config_vlans()
except Exception:
self.delete()
raise
else:
# no openstack, no need to create chains
-
if not config.l2_loopback and config.no_arp:
self._get_dest_macs_from_config()
if config.vlan_tagging:
if len(config.vlans) != 2:
raise ChainException('The config vlans property must be a list '
'with 2 lists of VLAN IDs')
- re_vlan = "[0-9]*$"
- self.vlans = [self._check_list('vlans[0]', config.vlans[0], re_vlan),
- self._check_list('vlans[1]', config.vlans[1], re_vlan)]
+ self._get_config_vlans()
if config.vxlan:
raise ChainException('VxLAN is only supported with OpenStack')
+ def _get_config_vlans(self):
+ re_vlan = "[0-9]*$"
+ try:
+ self.vlans = [self._check_list('vlans[0]', self.config.vlans[0], re_vlan),
+ self._check_list('vlans[1]', self.config.vlans[1], re_vlan)]
+ except IndexError:
+ raise ChainException('vlans parameter is mandatory. Set valid value in config file')
+
def _get_dest_macs_from_config(self):
re_mac = "[0-9a-fA-F]{2}([-:])[0-9a-fA-F]{2}(\\1[0-9a-fA-F]{2}){4}$"
tg_config = self.config.traffic_generator
port_index: left port is 0, right port is 1
return: a VLAN ID list indexed by the chain index or None if no vlan tagging
"""
- if self.chains:
+ if self.chains and self.is_admin:
return [self.chains[chain_index].get_vlan(port_index)
for chain_index in range(self.chain_count)]
# no openstack
from keystoneauth1.identity import v2
from keystoneauth1.identity import v3
from keystoneauth1 import session
+from keystoneclient import client
+from keystoneclient import utils
from log import LOG
self.rc_project_domain_name = None
self.rc_project_name = None
self.rc_identity_api_version = 2
+ self.is_admin = False
success = True
if openrc_file:
'Please enter your OpenStack Password: ')
if not self.rc_password:
self.rc_password = ""
+
+ # check if user has admin role in OpenStack project
+ try:
+ keystone = client.Client(session=self.get_session())
+ user = utils.find_resource(keystone.users, self.rc_username)
+ project = utils.find_resource(keystone.projects, self.rc_project_name)
+ roles = keystone.roles.list(user=user.id, project=project.id)
+ for role in roles:
+ if role.name == 'admin':
+ self.is_admin = True
+ except Exception:
+ LOG.warning("User is not admin, no permission to list user roles")
import pytest
from nfvbench.chain_runner import ChainRunner
+from nfvbench.chaining import ChainException
from nfvbench.chaining import ChainVnfPort
from nfvbench.chaining import InstancePlacer
from nfvbench.compute import Compute
openstack_spec = OpenStackSpec()
specs.set_openstack_spec(openstack_spec)
cred = MagicMock(spec=nfvbench.credentials.Credentials)
+ cred.is_admin = True
runner = ChainRunner(config, cred, specs, BasicFactory())
runner.close()
def test_pvp_chain_runner():
"""Test PVP chain runner."""
cred = MagicMock(spec=nfvbench.credentials.Credentials)
+ cred.is_admin = True
for shared_net in [True, False]:
for sc in [ChainType.PVP]:
for scc in [1, 2]:
config = _get_chain_config(sc, scc, shared_net)
_test_pvp_chain(config, cred)
+
+# Test not admin exception with empty value is raised
+@patch.object(Compute, 'find_image', _mock_find_image)
+@patch('nfvbench.chaining.Client')
+@patch('nfvbench.chaining.neutronclient')
+@patch('nfvbench.chaining.glanceclient')
+def _test_pvp_chain_no_admin_no_config_values(config, cred, mock_glance, mock_neutron, mock_client):
+ # instance = self.novaclient.servers.create(name=vmname,...)
+ # instance.status == 'ACTIVE'
+ mock_client.return_value.servers.create.return_value.status = 'ACTIVE'
+ netw = {'id': 0, 'provider:network_type': 'vlan', 'provider:segmentation_id': 1000}
+ mock_neutron.Client.return_value.create_network.return_value = {'network': netw}
+ mock_neutron.Client.return_value.list_networks.return_value = {'networks': None}
+ specs = Specs()
+ openstack_spec = OpenStackSpec()
+ specs.set_openstack_spec(openstack_spec)
+ runner = ChainRunner(config, cred, specs, BasicFactory())
+ runner.close()
+
+def test_pvp_chain_runner_no_admin_no_config_values():
+ """Test PVP chain runner."""
+ cred = MagicMock(spec=nfvbench.credentials.Credentials)
+ cred.is_admin = False
+ for shared_net in [True, False]:
+ for sc in [ChainType.PVP]:
+ for scc in [1, 2]:
+ config = _get_chain_config(sc, scc, shared_net)
+ with pytest.raises(ChainException):
+ _test_pvp_chain_no_admin_no_config_values(config, cred)
+
+# Test not admin with mandatory parameters values in config file
+@patch.object(Compute, 'find_image', _mock_find_image)
+@patch('nfvbench.chaining.Client')
+@patch('nfvbench.chaining.neutronclient')
+@patch('nfvbench.chaining.glanceclient')
+def _test_pvp_chain_no_admin_config_values(config, cred, mock_glance, mock_neutron, mock_client):
+ # instance = self.novaclient.servers.create(name=vmname,...)
+ # instance.status == 'ACTIVE'
+ mock_client.return_value.servers.create.return_value.status = 'ACTIVE'
+ netw = {'id': 0, 'provider:network_type': 'vlan', 'provider:segmentation_id': 1000}
+ mock_neutron.Client.return_value.create_network.return_value = {'network': netw}
+ mock_neutron.Client.return_value.list_networks.return_value = {'networks': None}
+ specs = Specs()
+ openstack_spec = OpenStackSpec()
+ specs.set_openstack_spec(openstack_spec)
+ runner = ChainRunner(config, cred, specs, BasicFactory())
+ runner.close()
+
+def test_pvp_chain_runner_no_admin_config_values():
+ """Test PVP chain runner."""
+ cred = MagicMock(spec=nfvbench.credentials.Credentials)
+ cred.is_admin = False
+ for shared_net in [True, False]:
+ for sc in [ChainType.PVP]:
+ for scc in [1, 2]:
+ config = _get_chain_config(sc, scc, shared_net)
+ config.availability_zone = "az"
+ config.hypervisor_hostname = "server"
+ # these are the 2 valid forms of vlan ranges
+ if scc == 1:
+ config.vlans = [100, 200]
+ else:
+ config.vlans = [[port * 100 + index for index in range(scc)]
+ for port in range(2)]
+ _test_pvp_chain_no_admin_config_values(config, cred)
+
+
@patch.object(Compute, 'find_image', _mock_find_image)
@patch('nfvbench.chaining.Client')
@patch('nfvbench.chaining.neutronclient')
openstack_spec = OpenStackSpec()
specs.set_openstack_spec(openstack_spec)
cred = MagicMock(spec=nfvbench.credentials.Credentials)
+ cred.is_admin = True
runner = ChainRunner(config, cred, specs, BasicFactory())
runner.close()
shared/not shared net x arp/no_arp x scc 1 or 2
"""
cred = MagicMock(spec=nfvbench.credentials.Credentials)
+ cred.is_admin = True
for shared_net in [True, False]:
for no_arp in [False, True]:
for scc in [1, 2]:
Code Review / nfvbench.git / commitdiff