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
from nfvbench.summarizer import _annotate_chain_stats
from nfvbench.traffic_client import TrafficClient
from nfvbench.traffic_gen.traffic_base import Latency
-from nfvbench.traffic_gen.trex import TRex
+from nfvbench.traffic_gen.trex_gen import TRex
# just to get rid of the unused function warning
def _mock_find_image(self, image_name):
- return True
+ return MagicMock()
@patch.object(Compute, 'find_image', _mock_find_image)
@patch('nfvbench.chaining.Client')
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]:
config = _get_chain_config(ChainType.EXT, scc, shared_net)
config.no_arp = no_arp
+ # this time use a tuple of network names
+ config['external_networks']['left'] = ('ext-lnet00', 'ext-lnet01')
+ config['external_networks']['right'] = ('ext-rnet00', 'ext-rnet01')
if no_arp:
# If EXT and no arp, the config must provide mac addresses (1 pair per chain)
config['traffic_generator']['mac_addrs_left'] = ['00:00:00:00:00:00'] * scc
if l2_loopback:
config.l2_loopback = True
config.vlans = [[100], [200]]
+ if sc == ChainType.EXT:
+ config['external_networks']['left'] = 'ext-lnet'
+ config['external_networks']['right'] = 'ext-rnet'
factory = BasicFactory()
config_plugin = factory.get_config_plugin_class()(config)
config = config_plugin.get_config()
@patch.object(Compute, 'find_image', _mock_find_image)
@patch.object(TrafficClient, 'skip_sleep', lambda x: True)
@patch.object(ChainVnfPort, 'get_mac', _mock_get_mac)
+@patch.object(TrafficClient, 'is_udp', lambda x, y: True)
@patch('nfvbench.chaining.Client')
@patch('nfvbench.chaining.neutronclient')
@patch('nfvbench.chaining.glanceclient')
@patch.object(Compute, 'find_image', _mock_find_image)
@patch.object(TrafficClient, 'skip_sleep', lambda x: True)
+@patch.object(TrafficClient, 'is_udp', lambda x, y: True)
@patch('nfvbench.chaining.Client')
@patch('nfvbench.chaining.neutronclient')
@patch('nfvbench.chaining.glanceclient')
@patch.object(Compute, 'find_image', _mock_find_image)
@patch.object(TrafficClient, 'skip_sleep', lambda x: True)
+@patch.object(TrafficClient, 'is_udp', lambda x, y: True)
@patch('nfvbench.chaining.Client')
@patch('nfvbench.chaining.neutronclient')
@patch('nfvbench.chaining.glanceclient')
'total': {'packets': [30000004, 30000004, 30000004, 30000004, 30000004, 30000004]}},
{0: {'packets': [15000002, '', 14000002, 14000002, '', 13000002]},
1: {'packets': [15000002, '', 15000002, 15000002, '', 15000002]},
- 'total': {'packets': [30000004, 29000004, 29000004, 29000004, 29000004, 28000004]}}]
+ 'total': {'packets': [30000004, 29000004, 29000004, 29000004, 29000004, 28000004]}},
+ # example with non-available rx count in last position
+ {0: {'packets': [2000054, 1999996, None]},
+ 1: {'packets': [2000054, 2000054, None]},
+ 'total': {'packets': [4000108, 4000050, 4000050]}}]
XP_CHAIN_STATS = [{0: {'packets': [2000054, '-58 (-0.0029%)', 1999996]}},
{0: {'packets': [2000054, '-58 (-0.0029%)', 1999996]},
1: {'packets': [2000054, '=>', 2000054]},
'-1,000,000 (-7.1429%)']},
1: {'packets': [15000002, '', '=>', '=>', '', 15000002]},
'total': {'packets': [30000004, '-1,000,000 (-3.3333%)', '=>', '=>', '=>',
- '-1,000,000 (-3.4483%)']}}]
+ '-1,000,000 (-3.4483%)']}},
+ {0: {'packets': [2000054, '-58 (-0.0029%)', 'n/a']},
+ 1: {'packets': [2000054, '=>', 'n/a']},
+ 'total': {'packets': [4000108, '-58 (-0.0014%)', 4000050]}}]
def test_summarizer():
assert stats == exp_stats
@patch.object(TrafficClient, 'skip_sleep', lambda x: True)
+@patch.object(TrafficClient, 'is_udp', lambda x, y: True)
def test_fixed_rate_no_openstack():
"""Test FIxed Rate run - no openstack."""
config = _get_chain_config(ChainType.EXT, 1, True, rate='100%')
Code Review / nfvbench.git / blobdiff