diff options
Diffstat (limited to 'test/test_chains.py')
| -rw-r--r-- | test/test_chains.py | 105 |
1 files changed, 99 insertions, 6 deletions
diff --git a/test/test_chains.py b/test/test_chains.py index ebc606e..d6f32c7 100644 --- a/test/test_chains.py +++ b/test/test_chains.py @@ -15,13 +15,14 @@ # """Test Chaining functions.""" -from mock_trex import no_op - from mock import MagicMock from mock import patch import pytest +from .mock_trex import no_op + 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 @@ -37,8 +38,8 @@ from nfvbench.specs import Specs 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 +from nfvbench import utils # just to get rid of the unused function warning no_op() @@ -69,6 +70,10 @@ def _get_chain_config(sc=ChainType.PVP, scc=1, shared_net=True, rate='1Mpps'): config.duration_sec = 2 config.interval_sec = 1 config.openrc_file = "dummy.rc" + config.no_flow_stats = False + config.no_latency_stats = False + config.no_latency_streams = False + config.loop_vm_arp = True return config def test_chain_runner_ext_no_openstack(): @@ -102,9 +107,13 @@ def test_chain_runner_ext_no_openstack(): def _mock_find_image(self, image_name): - return True + return MagicMock() + +def _mock_waiting_servers_deletion(nova_client, servers): + return MagicMock() @patch.object(Compute, 'find_image', _mock_find_image) +@patch.object(utils, 'waiting_servers_deletion', _mock_waiting_servers_deletion) @patch('nfvbench.chaining.Client') @patch('nfvbench.chaining.neutronclient') @patch('nfvbench.chaining.glanceclient') @@ -119,18 +128,89 @@ def _test_pvp_chain(config, cred, mock_glance, mock_neutron, mock_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.object(utils, 'waiting_servers_deletion', _mock_waiting_servers_deletion) +@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/mock 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.object(utils, 'waiting_servers_deletion', _mock_waiting_servers_deletion) +@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') @@ -145,6 +225,7 @@ def _test_ext_chain(config, cred, mock_glance, mock_neutron, mock_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() @@ -155,11 +236,15 @@ def test_ext_chain_runner(): 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 @@ -172,6 +257,9 @@ def _check_nfvbench_openstack(sc=ChainType.PVP, l2_loopback=False): 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() @@ -179,7 +267,7 @@ def _check_nfvbench_openstack(sc=ChainType.PVP, l2_loopback=False): nfvb = NFVBench(config, openstack_spec, config_plugin, factory) res = nfvb.run({}, 'pytest') if res['status'] != 'OK': - print res + print(res) assert res['status'] == 'OK' @@ -193,6 +281,8 @@ def _mock_get_mac(dummy): @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.object(utils, 'waiting_servers_deletion', _mock_waiting_servers_deletion) @patch('nfvbench.chaining.Client') @patch('nfvbench.chaining.neutronclient') @patch('nfvbench.chaining.glanceclient') @@ -209,6 +299,7 @@ def test_nfvbench_run(mock_cred, mock_glance, mock_neutron, mock_client): @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') @@ -224,6 +315,7 @@ def test_nfvbench_ext_arp(mock_cred, mock_glance, mock_neutron, mock_client): @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') @@ -388,6 +480,7 @@ 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%') |