diff options
Diffstat (limited to 'networking_sfc/tests/unit/db/test_sfc_db.py')
| -rw-r--r-- | networking_sfc/tests/unit/db/test_sfc_db.py | 1490 |
1 files changed, 1490 insertions, 0 deletions
diff --git a/networking_sfc/tests/unit/db/test_sfc_db.py b/networking_sfc/tests/unit/db/test_sfc_db.py new file mode 100644 index 0000000..70d57e3 --- /dev/null +++ b/networking_sfc/tests/unit/db/test_sfc_db.py @@ -0,0 +1,1490 @@ +# Copyright 2015 Futurewei. All rights reserved. +# +# 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 contextlib +import mock +import six +import webob.exc + +from oslo_config import cfg +from oslo_utils import importutils +from oslo_utils import uuidutils + +from neutron.api import extensions as api_ext +from neutron.common import config +import neutron.extensions as nextensions + +from networking_sfc.db import flowclassifier_db as fdb +from networking_sfc.db import sfc_db +from networking_sfc import extensions +from networking_sfc.extensions import flowclassifier as fc_ext +from networking_sfc.extensions import sfc +from networking_sfc.tests import base +from networking_sfc.tests.unit.db import test_flowclassifier_db + + +DB_SFC_PLUGIN_CLASS = ( + "networking_sfc.db.sfc_db.SfcDbPlugin" +) +extensions_path = ':'.join(extensions.__path__ + nextensions.__path__) + + +class SfcDbPluginTestCaseBase( + base.BaseTestCase +): + def _create_port_chain( + self, fmt, port_chain=None, expected_res_status=None, **kwargs + ): + ctx = kwargs.get('context', None) + tenant_id = kwargs.get('tenant_id', self._tenant_id) + data = {'port_chain': port_chain or {}} + if ctx is None: + data['port_chain'].update({'tenant_id': tenant_id}) + req = self.new_create_request( + 'port_chains', data, fmt, context=ctx + ) + res = req.get_response(self.ext_api) + if expected_res_status: + self.assertEqual(res.status_int, expected_res_status) + return res + + @contextlib.contextmanager + def port_chain(self, fmt=None, port_chain=None, do_delete=True, **kwargs): + if not fmt: + fmt = self.fmt + res = self._create_port_chain(fmt, port_chain, **kwargs) + if res.status_int >= 400: + raise webob.exc.HTTPClientError(code=res.status_int) + port_chain = self.deserialize(fmt or self.fmt, res) + yield port_chain + if do_delete: + self._delete('port_chains', port_chain['port_chain']['id']) + + def _create_port_pair_group( + self, fmt, port_pair_group=None, expected_res_status=None, **kwargs + ): + ctx = kwargs.get('context', None) + tenant_id = kwargs.get('tenant_id', self._tenant_id) + data = {'port_pair_group': port_pair_group or {}} + if ctx is None: + data['port_pair_group'].update({'tenant_id': tenant_id}) + req = self.new_create_request( + 'port_pair_groups', data, fmt, context=ctx + ) + res = req.get_response(self.ext_api) + if expected_res_status: + self.assertEqual(res.status_int, expected_res_status) + return res + + @contextlib.contextmanager + def port_pair_group( + self, fmt=None, port_pair_group=None, do_delete=True, **kwargs + ): + if not fmt: + fmt = self.fmt + res = self._create_port_pair_group(fmt, port_pair_group, **kwargs) + if res.status_int >= 400: + raise webob.exc.HTTPClientError(code=res.status_int) + port_pair_group = self.deserialize(fmt or self.fmt, res) + yield port_pair_group + if do_delete: + self._delete( + 'port_pair_groups', + port_pair_group['port_pair_group']['id']) + + def _create_port_pair( + self, fmt, port_pair=None, expected_res_status=None, **kwargs + ): + ctx = kwargs.get('context', None) + tenant_id = kwargs.get('tenant_id', self._tenant_id) + data = {'port_pair': port_pair or {}} + if ctx is None: + data['port_pair'].update({'tenant_id': tenant_id}) + req = self.new_create_request( + 'port_pairs', data, fmt, context=ctx + ) + res = req.get_response(self.ext_api) + if expected_res_status: + self.assertEqual(res.status_int, expected_res_status) + return res + + @contextlib.contextmanager + def port_pair(self, fmt=None, port_pair=None, do_delete=True, **kwargs): + if not fmt: + fmt = self.fmt + res = self._create_port_pair(fmt, port_pair, **kwargs) + if res.status_int >= 400: + raise webob.exc.HTTPClientError(code=res.status_int) + port_pair = self.deserialize(fmt or self.fmt, res) + yield port_pair + if do_delete: + self._delete('port_pairs', port_pair['port_pair']['id']) + + def _get_expected_port_pair(self, port_pair): + return { + 'name': port_pair.get('name') or '', + 'description': port_pair.get('description') or '', + 'egress': port_pair.get('egress'), + 'ingress': port_pair.get('ingress'), + 'service_function_parameters': port_pair.get( + 'service_function_parameters') or {'correlation': None} + } + + def _test_create_port_pair(self, port_pair, expected_port_pair=None): + if expected_port_pair is None: + expected_port_pair = self._get_expected_port_pair(port_pair) + with self.port_pair(port_pair=port_pair) as pp: + for k, v in six.iteritems(expected_port_pair): + self.assertEqual(pp['port_pair'][k], v) + + def _test_create_port_pairs( + self, port_pairs, expected_port_pairs=None + ): + if port_pairs: + port_pair = port_pairs.pop() + if expected_port_pairs: + expected_port_pair = expected_port_pairs.pop() + else: + expected_port_pair = self._get_expected_port_pair(port_pair) + with self.port_pair(port_pair=port_pair) as pp: + for k, v in six.iteritems(expected_port_pair): + self.assertEqual(pp['port_pair'][k], v) + + def _get_expected_port_pair_group(self, port_pair_group): + return { + 'name': port_pair_group.get('name') or '', + 'description': port_pair_group.get('description') or '', + 'port_pairs': port_pair_group.get('port_pairs') or [] + } + + def _test_create_port_pair_group( + self, port_pair_group, expected_port_pair_group=None + ): + if expected_port_pair_group is None: + expected_port_pair_group = self._get_expected_port_pair_group( + port_pair_group) + with self.port_pair_group(port_pair_group=port_pair_group) as pg: + for k, v in six.iteritems(expected_port_pair_group): + self.assertEqual(pg['port_pair_group'][k], v) + + def _test_create_port_pair_groups( + self, port_pair_groups, expected_port_pair_groups=None + ): + if port_pair_groups: + port_pair_group = port_pair_groups.pop() + if expected_port_pair_groups: + expected_port_pair_group = expected_port_pair_groups.pop() + else: + expected_port_pair_group = self._get_expected_port_pair_group( + port_pair_group) + with self.port_pair_group(port_pair_group=port_pair_group) as pg: + for k, v in six.iteritems(expected_port_pair_group): + self.assertEqual(pg['port_pair_group'][k], v) + + def _get_expected_port_chain(self, port_chain): + return { + 'name': port_chain.get('name') or '', + 'description': port_chain.get('description') or '', + 'port_pair_groups': port_chain['port_pair_groups'], + 'flow_classifiers': port_chain.get('flow_classifiers') or [], + 'chain_parameters': port_chain.get( + 'chain_parameters') or {'correlation': 'mpls'} + } + + def _test_create_port_chain(self, port_chain, expected_port_chain=None): + if expected_port_chain is None: + expected_port_chain = self._get_expected_port_chain(port_chain) + with self.port_chain(port_chain=port_chain) as pc: + for k, v in six.iteritems(expected_port_chain): + self.assertEqual(pc['port_chain'][k], v) + + def _test_create_port_chains( + self, port_chains, expected_port_chains=None + ): + if port_chains: + port_chain = port_chains.pop() + if expected_port_chains: + expected_port_chain = expected_port_chains.pop() + else: + expected_port_chain = self._get_expected_port_chain( + port_chain) + with self.port_chain(port_chain=port_chain) as pc: + for k, v in six.iteritems(expected_port_chain): + self.assertEqual(pc['port_chain'][k], v) + + +class SfcDbPluginTestCase( + base.NeutronDbPluginV2TestCase, + test_flowclassifier_db.FlowClassifierDbPluginTestCaseBase, + SfcDbPluginTestCaseBase +): + resource_prefix_map = dict([ + (k, sfc.SFC_PREFIX) + for k in sfc.RESOURCE_ATTRIBUTE_MAP.keys() + ] + [ + (k, fc_ext.FLOW_CLASSIFIER_PREFIX) + for k in fc_ext.RESOURCE_ATTRIBUTE_MAP.keys() + ]) + + def setUp(self, core_plugin=None, sfc_plugin=None, + flowclassifier_plugin=None, ext_mgr=None): + mock_log_p = mock.patch.object(sfc_db, 'LOG') + self.mock_log = mock_log_p.start() + cfg.CONF.register_opts(sfc.sfc_quota_opts, 'QUOTAS') + if not sfc_plugin: + sfc_plugin = DB_SFC_PLUGIN_CLASS + if not flowclassifier_plugin: + flowclassifier_plugin = ( + test_flowclassifier_db.DB_FLOWCLASSIFIER_PLUGIN_CLASS) + + service_plugins = { + sfc.SFC_EXT: sfc_plugin, + fc_ext.FLOW_CLASSIFIER_EXT: flowclassifier_plugin + } + sfc_db.SfcDbPlugin.supported_extension_aliases = [ + "sfc"] + sfc_db.SfcDbPlugin.path_prefix = sfc.SFC_PREFIX + fdb.FlowClassifierDbPlugin.supported_extension_aliases = [ + "flow_classifier"] + fdb.FlowClassifierDbPlugin.path_prefix = ( + fc_ext.FLOW_CLASSIFIER_PREFIX + ) + super(SfcDbPluginTestCase, self).setUp( + ext_mgr=ext_mgr, + plugin=core_plugin, + service_plugins=service_plugins + ) + if not ext_mgr: + self.sfc_plugin = importutils.import_object(sfc_plugin) + self.flowclassifier_plugin = importutils.import_object( + flowclassifier_plugin) + ext_mgr = api_ext.PluginAwareExtensionManager( + extensions_path, + { + sfc.SFC_EXT: self.sfc_plugin, + fc_ext.FLOW_CLASSIFIER_EXT: self.flowclassifier_plugin + } + ) + app = config.load_paste_app('extensions_test_app') + self.ext_api = api_ext.ExtensionMiddleware(app, ext_mgr=ext_mgr) + + def test_create_port_chain(self): + with self.port_pair_group(port_pair_group={}) as pg: + self._test_create_port_chain({ + 'port_pair_groups': [pg['port_pair_group']['id']]}) + + def test_quota_create_port_chain(self): + cfg.CONF.set_override('quota_port_chain', 3, group='QUOTAS') + with self.port_pair_group( + port_pair_group={}, do_delete=False + ) as pg1, self.port_pair_group( + port_pair_group={}, do_delete=False + ) as pg2, self.port_pair_group( + port_pair_group={}, do_delete=False + ) as pg3, self.port_pair_group( + port_pair_group={}, do_delete=False + ) as pg4: + self._create_port_chain( + self.fmt, { + 'port_pair_groups': [pg1['port_pair_group']['id']] + }, expected_res_status=201) + self._create_port_chain( + self.fmt, { + 'port_pair_groups': [pg2['port_pair_group']['id']] + }, expected_res_status=201) + self._create_port_chain( + self.fmt, { + 'port_pair_groups': [pg3['port_pair_group']['id']] + }, expected_res_status=201) + self._create_port_chain( + self.fmt, { + 'port_pair_groups': [pg4['port_pair_group']['id']] + }, expected_res_status=409) + + def test_create_port_chain_all_fields(self): + with self.port_pair_group(port_pair_group={}) as pg: + self._test_create_port_chain({ + 'port_pair_groups': [pg['port_pair_group']['id']], + 'flow_classifiers': [], + 'name': 'abc', + 'description': 'def', + 'chain_parameters': {'correlation': 'mpls'} + }) + + def test_create_port_chain_multi_port_pair_groups(self): + with self.port_pair_group( + port_pair_group={} + ) as pg1, self.port_pair_group( + port_pair_group={} + ) as pg2: + self._test_create_port_chain({ + 'port_pair_groups': [ + pg1['port_pair_group']['id'], + pg2['port_pair_group']['id'] + ] + }) + + def test_create_port_chain_shared_port_pair_groups(self): + with self.port_pair_group( + port_pair_group={} + ) as pg1, self.port_pair_group( + port_pair_group={} + ) as pg2, self.port_pair_group( + port_pair_group={} + ) as pg3: + self._test_create_port_chains([{ + 'port_pair_groups': [ + pg1['port_pair_group']['id'], + pg2['port_pair_group']['id'] + ] + }, { + 'port_pair_groups': [ + pg1['port_pair_group']['id'], + pg3['port_pair_group']['id'] + ] + }]) + + def test_create_port_chain_shared_port_pair_groups_different_order(self): + with self.port_pair_group( + port_pair_group={} + ) as pg1, self.port_pair_group( + port_pair_group={} + ) as pg2: + self._test_create_port_chains([{ + 'port_pair_groups': [ + pg1['port_pair_group']['id'], + pg2['port_pair_group']['id'] + ] + }, { + 'port_pair_groups': [ + pg2['port_pair_group']['id'], + pg1['port_pair_group']['id'] + ] + }]) + + def test_create_port_chain_with_empty_chain_parameters(self): + with self.port_pair_group(port_pair_group={}) as pg: + self._test_create_port_chain({ + 'chain_parameters': {}, + 'port_pair_groups': [pg['port_pair_group']['id']] + }) + + def test_create_port_chain_with_none_chain_parameters(self): + with self.port_pair_group(port_pair_group={}) as pg: + self._test_create_port_chain({ + 'chain_parameters': None, + 'port_pair_groups': [pg['port_pair_group']['id']] + }) + + def test_create_port_chain_with_default_chain_parameters(self): + with self.port_pair_group(port_pair_group={}) as pg: + self._test_create_port_chain({ + 'chain_parameters': {'correlation': 'mpls'}, + 'port_pair_groups': [pg['port_pair_group']['id']] + }) + + def test_create_port_chain_with_none_flow_classifiers(self): + with self.port_pair_group(port_pair_group={}) as pg: + self._test_create_port_chain({ + 'flow_classifiers': None, + 'port_pair_groups': [pg['port_pair_group']['id']] + }) + + def test_create_port_chain_with_empty_flow_classifiers(self): + with self.port_pair_group(port_pair_group={}) as pg: + self._test_create_port_chain({ + 'flow_classifiers': [], + 'port_pair_groups': [pg['port_pair_group']['id']] + }) + + def test_create_port_chain_with_flow_classifiers(self): + with self.flow_classifier(flow_classifier={}) as fc: + with self.port_pair_group(port_pair_group={}) as pg: + self._test_create_port_chain({ + 'flow_classifiers': [fc['flow_classifier']['id']], + 'port_pair_groups': [pg['port_pair_group']['id']] + }) + + def test_create_port_chain_with_multi_flow_classifiers(self): + with self.flow_classifier( + flow_classifier={} + ) as fc1, self.flow_classifier( + flow_classifier={} + ) as fc2: + with self.port_pair_group(port_pair_group={}) as pg: + self._test_create_port_chain({ + 'flow_classifiers': [ + fc1['flow_classifier']['id'], + fc2['flow_classifier']['id'] + ], + 'port_pair_groups': [pg['port_pair_group']['id']] + }) + + def test_create_port_chain_with_port_pairs(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pp1, self.port_pair(port_pair={ + 'ingress': dst_port['port']['id'], + 'egress': src_port['port']['id'] + }) as pp2: + with self.port_pair_group(port_pair_group={ + 'port_pairs': [ + pp1['port_pair']['id'] + ] + }) as pg1, self.port_pair_group(port_pair_group={ + 'port_pairs': [ + pp2['port_pair']['id'] + ] + }) as pg2: + self._test_create_port_chain({ + 'port_pair_groups': [ + pg1['port_pair_group']['id'], + pg2['port_pair_group']['id'] + ] + }) + + def test_create_port_chain_with_empty_port_pair_groups(self): + self._create_port_chain( + self.fmt, {'port_pair_groups': []}, + expected_res_status=400 + ) + + def test_create_port_chain_with_nonuuid_port_pair_group_id(self): + self._create_port_chain( + self.fmt, {'port_pair_groups': ['unknown']}, + expected_res_status=400 + ) + + def test_create_port_chain_with_unknown_port_pair_group_id(self): + self._create_port_chain( + self.fmt, {'port_pair_groups': [uuidutils.generate_uuid()]}, + expected_res_status=404 + ) + + def test_create_port_chain_with_same_port_pair_groups(self): + with self.port_pair_group( + port_pair_group={} + ) as pg: + with self.port_chain( + port_chain={ + 'port_pair_groups': [pg['port_pair_group']['id']] + } + ): + self._create_port_chain( + self.fmt, { + 'port_pair_groups': [pg['port_pair_group']['id']] + }, expected_res_status=409 + ) + + def test_create_port_chain_with_no_port_pair_groups(self): + self._create_port_chain( + self.fmt, {}, expected_res_status=400 + ) + + def test_create_port_chain_with_invalid_chain_parameters(self): + with self.port_pair_group(port_pair_group={}) as pg: + self._create_port_chain( + self.fmt, { + 'chain_parameters': {'correlation': 'unknown'}, + 'port_pair_groups': [pg['port_pair_group']['id']] + }, expected_res_status=400 + ) + + def test_create_port_chain_unknown_flow_classifiers(self): + with self.port_pair_group(port_pair_group={}) as pg: + self._create_port_chain( + self.fmt, { + 'flow_classifiers': [uuidutils.generate_uuid()], + 'port_pair_groups': [pg['port_pair_group']['id']] + }, expected_res_status=404 + ) + + def test_create_port_chain_nouuid_flow_classifiers(self): + with self.port_pair_group(port_pair_group={}) as pg: + self._create_port_chain( + self.fmt, { + 'flow_classifiers': ['unknown'], + 'port_pair_groups': [pg['port_pair_group']['id']] + }, expected_res_status=400 + ) + + def test_list_port_chains(self): + with self.port_pair_group( + port_pair_group={} + ) as pg1, self.port_pair_group( + port_pair_group={} + ) as pg2: + with self.port_chain(port_chain={ + 'port_pair_groups': [pg1['port_pair_group']['id']] + }) as pc1, self.port_chain(port_chain={ + 'port_pair_groups': [pg2['port_pair_group']['id']] + }) as pc2: + port_chains = [pc1, pc2] + self._test_list_resources( + 'port_chain', port_chains + ) + + def test_list_port_chains_with_params(self): + with self.port_pair_group( + port_pair_group={} + ) as pg1, self.port_pair_group( + port_pair_group={} + ) as pg2: + with self.port_chain(port_chain={ + 'name': 'test1', + 'port_pair_groups': [pg1['port_pair_group']['id']] + }) as pc1, self.port_chain(port_chain={ + 'name': 'test2', + 'port_pair_groups': [pg2['port_pair_group']['id']] + }) as pc2: + self._test_list_resources( + 'port_chain', [pc1], + query_params='name=test1' + ) + self._test_list_resources( + 'port_chain', [pc2], + query_params='name=test2' + ) + self._test_list_resources( + 'port_chain', [], + query_params='name=test3' + ) + + def test_list_port_chains_with_unknown_params(self): + with self.port_pair_group( + port_pair_group={} + ) as pg1, self.port_pair_group( + port_pair_group={} + ) as pg2: + with self.port_chain(port_chain={ + 'name': 'test1', + 'port_pair_groups': [pg1['port_pair_group']['id']] + }) as pc1, self.port_chain(port_chain={ + 'name': 'test2', + 'port_pair_groups': [pg2['port_pair_group']['id']] + }) as pc2: + self._test_list_resources( + 'port_chain', [pc1, pc2], + query_params='hello=test3' + ) + + def test_show_port_chain(self): + with self.port_pair_group( + port_pair_group={} + ) as pg: + with self.port_chain(port_chain={ + 'name': 'test1', + 'description': 'portchain', + 'port_pair_groups': [pg['port_pair_group']['id']] + }) as pc: + req = self.new_show_request( + 'port_chains', pc['port_chain']['id'] + ) + res = self.deserialize( + self.fmt, req.get_response(self.ext_api) + ) + expected = self._get_expected_port_chain(pc['port_chain']) + for k, v in six.iteritems(expected): + self.assertEqual(res['port_chain'][k], v) + + def test_show_port_chain_noexist(self): + req = self.new_show_request( + 'port_chains', '1' + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 404) + + def test_update_port_chain(self): + with self.flow_classifier( + flow_classifier={} + ) as fc1, self.flow_classifier( + flow_classifier={} + ) as fc2: + with self.port_pair_group( + port_pair_group={} + ) as pg: + with self.port_chain(port_chain={ + 'name': 'test1', + 'description': 'desc1', + 'port_pair_groups': [pg['port_pair_group']['id']], + 'flow_classifiers': [fc1['flow_classifier']['id']] + }) as pc: + updates = { + 'name': 'test2', + 'description': 'desc2', + 'flow_classifiers': [fc2['flow_classifier']['id']] + } + req = self.new_update_request( + 'port_chains', {'port_chain': updates}, + pc['port_chain']['id'] + ) + res = self.deserialize( + self.fmt, + req.get_response(self.ext_api) + ) + expected = pc['port_chain'] + expected.update(updates) + for k, v in six.iteritems(expected): + self.assertEqual(res['port_chain'][k], v) + req = self.new_show_request( + 'port_chains', pc['port_chain']['id'] + ) + res = self.deserialize( + self.fmt, req.get_response(self.ext_api) + ) + for k, v in six.iteritems(expected): + self.assertEqual(res['port_chain'][k], v) + + def test_update_port_chain_port_pair_groups(self): + with self.port_pair_group( + port_pair_group={} + ) as pg1, self.port_pair_group( + port_pair_group={} + ) as pg2: + with self.port_chain(port_chain={ + 'port_pair_groups': [pg1['port_pair_group']['id']], + }) as pc: + updates = { + 'port_pair_groups': [pg2['port_pair_group']['id']] + } + req = self.new_update_request( + 'port_chains', {'port_chain': updates}, + pc['port_chain']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 400) + + def test_update_port_chain_chain_parameters(self): + with self.port_pair_group( + port_pair_group={} + ) as pg: + with self.port_chain(port_chain={ + 'port_pair_groups': [pg['port_pair_group']['id']], + }) as pc: + updates = { + 'chain_parameters': {'correlation': 'mpls'} + } + req = self.new_update_request( + 'port_chains', {'port_chain': updates}, + pc['port_chain']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 400) + + def test_delete_port_chain(self): + with self.port_pair_group( + port_pair_group={} + ) as pg: + with self.port_chain(port_chain={ + 'port_pair_groups': [pg['port_pair_group']['id']] + }, do_delete=False) as pc: + req = self.new_delete_request( + 'port_chains', pc['port_chain']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 204) + req = self.new_show_request( + 'port_chains', pc['port_chain']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 404) + req = self.new_show_request( + 'port_pair_groups', pg['port_pair_group']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 200) + + def test_delete_port_chain_noexist(self): + req = self.new_delete_request( + 'port_chains', '1' + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 404) + + def test_delete_flow_classifier_port_chain_exist(self): + with self.flow_classifier(flow_classifier={ + }) as fc: + with self.port_pair_group(port_pair_group={ + }) as pg: + with self.port_chain(port_chain={ + 'port_pair_groups': [pg['port_pair_group']['id']], + 'flow_classifiers': [fc['flow_classifier']['id']] + }): + req = self.new_delete_request( + 'flow_classifiers', fc['flow_classifier']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 409) + + def test_create_port_pair_group(self): + self._test_create_port_pair_group({}) + + def test_quota_create_port_pair_group_quota(self): + cfg.CONF.set_override('quota_port_pair_group', 3, group='QUOTAS') + self._create_port_pair_group( + self.fmt, {'port_pairs': []}, expected_res_status=201 + ) + self._create_port_pair_group( + self.fmt, {'port_pairs': []}, expected_res_status=201 + ) + self._create_port_pair_group( + self.fmt, {'port_pairs': []}, expected_res_status=201 + ) + self._create_port_pair_group( + self.fmt, {'port_pairs': []}, expected_res_status=409 + ) + + def test_create_port_pair_group_all_fields(self): + self._test_create_port_pair_group({ + 'name': 'test1', + 'description': 'desc1', + 'port_pairs': [] + }) + + def test_create_port_pair_group_with_port_pairs(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pp1, self.port_pair(port_pair={ + 'ingress': dst_port['port']['id'], + 'egress': src_port['port']['id'] + }) as pp2: + self._test_create_port_pair_group({ + 'port_pairs': [ + pp1['port_pair']['id'], + pp2['port_pair']['id'] + ] + }) + + def test_create_port_pair_group_with_nouuid_port_pair_id(self): + self._create_port_pair_group( + self.fmt, {'port_pairs': ['unknown']}, + expected_res_status=400 + ) + + def test_create_port_pair_group_with_unknown_port_pair_id(self): + self._create_port_pair_group( + self.fmt, {'port_pairs': [uuidutils.generate_uuid()]}, + expected_res_status=404 + ) + + def test_create_port_pair_group_share_port_pair_id(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pp: + with self.port_pair_group(port_pair_group={ + 'port_pairs': [pp['port_pair']['id']] + }): + self._create_port_pair_group( + self.fmt, {'port_pairs': [pp['port_pair']['id']]}, + expected_res_status=409 + ) + + def test_list_port_pair_groups(self): + with self.port_pair_group(port_pair_group={ + 'name': 'test1' + }) as pc1, self.port_pair_group(port_pair_group={ + 'name': 'test2' + }) as pc2: + port_pair_groups = [pc1, pc2] + self._test_list_resources( + 'port_pair_group', port_pair_groups + ) + + def test_list_port_pair_groups_with_params(self): + with self.port_pair_group(port_pair_group={ + 'name': 'test1' + }) as pc1, self.port_pair_group(port_pair_group={ + 'name': 'test2' + }) as pc2: + self._test_list_resources( + 'port_pair_group', [pc1], + query_params='name=test1' + ) + self._test_list_resources( + 'port_pair_group', [pc2], + query_params='name=test2' + ) + self._test_list_resources( + 'port_pair_group', [], + query_params='name=test3' + ) + + def test_list_port_pair_groups_with_unknown_params(self): + with self.port_pair_group(port_pair_group={ + 'name': 'test1' + }) as pc1, self.port_pair_group(port_pair_group={ + 'name': 'test2' + }) as pc2: + self._test_list_resources( + 'port_pair_group', [pc1, pc2], + query_params='hello=test3' + ) + + def test_show_port_pair_group(self): + with self.port_pair_group(port_pair_group={ + 'name': 'test1' + }) as pc: + req = self.new_show_request( + 'port_pair_groups', pc['port_pair_group']['id'] + ) + res = self.deserialize( + self.fmt, req.get_response(self.ext_api) + ) + for k, v in six.iteritems(pc['port_pair_group']): + self.assertEqual(res['port_pair_group'][k], v) + + def test_show_port_pair_group_noexist(self): + req = self.new_show_request( + 'port_pair_groups', '1' + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 404) + + def test_update_port_pair_group(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pp1, self.port_pair(port_pair={ + 'ingress': dst_port['port']['id'], + 'egress': src_port['port']['id'] + }) as pp2: + with self.port_pair_group(port_pair_group={ + 'name': 'test1', + 'description': 'desc1', + 'port_pairs': [pp1['port_pair']['id']] + }) as pg: + updates = { + 'name': 'test2', + 'description': 'desc2', + 'port_pairs': [pp2['port_pair']['id']] + } + req = self.new_update_request( + 'port_pair_groups', {'port_pair_group': updates}, + pg['port_pair_group']['id'] + ) + res = self.deserialize( + self.fmt, + req.get_response(self.ext_api) + ) + expected = pg['port_pair_group'] + expected.update(updates) + for k, v in six.iteritems(expected): + self.assertEqual(res['port_pair_group'][k], v) + req = self.new_show_request( + 'port_pair_groups', pg['port_pair_group']['id'] + ) + res = self.deserialize( + self.fmt, req.get_response(self.ext_api) + ) + for k, v in six.iteritems(expected): + self.assertEqual(res['port_pair_group'][k], v) + + def test_delete_port_pair_group(self): + with self.port_pair_group(port_pair_group={ + 'name': 'test1' + }, do_delete=False) as pc: + req = self.new_delete_request( + 'port_pair_groups', pc['port_pair_group']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 204) + req = self.new_show_request( + 'port_pair_groups', pc['port_pair_group']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 404) + + def test_delete_port_pair_group_port_chain_exist(self): + with self.port_pair_group(port_pair_group={ + 'name': 'test1' + }) as pg: + with self.port_chain(port_chain={ + 'port_pair_groups': [pg['port_pair_group']['id']] + }): + req = self.new_delete_request( + 'port_pair_groups', pg['port_pair_group']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 409) + + def test_delete_port_pair_group_noexist(self): + req = self.new_delete_request( + 'port_pair_groups', '1' + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 404) + + def test_create_port_pair(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + self._test_create_port_pair({ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) + + def test_quota_create_port_pair_quota(self): + cfg.CONF.set_override('quota_port_pair', 3, group='QUOTAS') + with self.port( + name='port1', + device_id='default' + ) as src_port1, self.port( + name='port2', + device_id='default' + ) as dst_port1, self.port( + name='port3', + device_id='default' + ) as src_port2, self.port( + name='port4', + device_id='default' + ) as dst_port2, self.port( + name='port5', + device_id='default' + ) as src_port3, self.port( + name='port6', + device_id='default' + ) as dst_port3, self.port( + name='port7', + device_id='default' + ) as src_port4, self.port( + name='port8', + device_id='default' + ) as dst_port4: + self._create_port_pair( + self.fmt, { + 'ingress': src_port1['port']['id'], + 'egress': dst_port1['port']['id'] + }, expected_res_status=201) + self._create_port_pair( + self.fmt, { + 'ingress': src_port2['port']['id'], + 'egress': dst_port2['port']['id'] + }, expected_res_status=201) + self._create_port_pair( + self.fmt, { + 'ingress': src_port3['port']['id'], + 'egress': dst_port3['port']['id'] + }, expected_res_status=201) + self._create_port_pair( + self.fmt, { + 'ingress': src_port4['port']['id'], + 'egress': dst_port4['port']['id'] + }, expected_res_status=409) + + def test_create_port_pair_all_fields(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + self._test_create_port_pair({ + 'name': 'test1', + 'description': 'desc1', + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'], + 'service_function_parameters': {'correlation': None} + }) + + def test_create_port_pair_none_service_function_parameters(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + self._test_create_port_pair({ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'], + 'service_function_parameters': None + }) + + def test_create_port_pair_empty_service_function_parameters(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + self._test_create_port_pair({ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'], + 'service_function_parameters': {} + }) + + def test_create_port_pair_with_src_dst_same_port(self): + with self.port( + name='port1', + device_id='default' + ) as src_dst_port: + self._test_create_port_pair({ + 'ingress': src_dst_port['port']['id'], + 'egress': src_dst_port['port']['id'] + }) + + def test_create_port_pair_empty_input(self): + self._create_port_pair(self.fmt, {}, expected_res_status=400) + + def test_create_port_pair_with_no_ingress(self): + with self.port( + name='port1', + device_id='default' + ) as dst_port: + self._create_port_pair( + self.fmt, + { + 'egress': dst_port['port']['id'] + }, + expected_res_status=400 + ) + + def test_create_port_pair_with_no_egress(self): + with self.port( + name='port1', + device_id='default' + ) as src_port: + self._create_port_pair( + self.fmt, + { + 'ingress': src_port['port']['id'] + }, + expected_res_status=400 + ) + + def test_create_port_pair_with_nouuid_ingress(self): + with self.port( + name='port1', + device_id='default' + ) as dst_port: + self._create_port_pair( + self.fmt, + { + 'ingress': '1', + 'egress': dst_port['port']['id'] + }, + expected_res_status=400 + ) + + def test_create_port_pair_with_unknown_ingress(self): + with self.port( + name='port1', + device_id='default' + ) as dst_port: + self._create_port_pair( + self.fmt, + { + 'ingress': uuidutils.generate_uuid(), + 'egress': dst_port['port']['id'] + }, + expected_res_status=404 + ) + + def test_create_port_pair_with_nouuid_egress(self): + with self.port( + name='port1', + device_id='default' + ) as src_port: + self._create_port_pair( + self.fmt, + { + 'ingress': src_port['port']['id'], + 'egress': '1' + }, + expected_res_status=400 + ) + + def test_create_port_pair_with_unkown_egress(self): + with self.port( + name='port1', + device_id='default' + ) as src_port: + self._create_port_pair( + self.fmt, + { + 'ingress': src_port['port']['id'], + 'egress': uuidutils.generate_uuid() + }, + expected_res_status=404 + ) + + def test_create_port_pair_ingress_egress_different_hosts(self): + with self.port( + name='port1', + device_id='device1' + ) as src_port, self.port( + name='port2', + device_id='device2' + ) as dst_port: + self._create_port_pair( + self.fmt, + { + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }, + expected_res_status=400 + ) + + def test_create_port_pair_with_invalid_service_function_parameters(self): + with self.port( + name='port1', + device_id='default' + ) as src_dst_port: + self._create_port_pair( + self.fmt, + { + 'ingress': src_dst_port['port']['id'], + 'egress': src_dst_port['port']['id'], + 'service_function_parameters': {'abc': 'def'} + }, + expected_res_status=400 + ) + + def test_list_port_pairs(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pc1, self.port_pair(port_pair={ + 'ingress': dst_port['port']['id'], + 'egress': src_port['port']['id'] + }) as pc2: + port_pairs = [pc1, pc2] + self._test_list_resources( + 'port_pair', port_pairs + ) + + def test_list_port_pairs_with_params(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'name': 'test1', + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pc1, self.port_pair(port_pair={ + 'name': 'test2', + 'ingress': dst_port['port']['id'], + 'egress': src_port['port']['id'] + }) as pc2: + self._test_list_resources( + 'port_pair', [pc1], + query_params='name=test1' + ) + self._test_list_resources( + 'port_pair', [pc2], + query_params='name=test2' + ) + self._test_list_resources( + 'port_pair', [], + query_params='name=test3' + ) + + def test_list_port_pairs_with_unknown_params(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'name': 'test1', + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pc1, self.port_pair(port_pair={ + 'name': 'test2', + 'ingress': dst_port['port']['id'], + 'egress': src_port['port']['id'] + }) as pc2: + port_pairs = [pc1, pc2] + self._test_list_resources( + 'port_pair', port_pairs, + query_params='hello=test3' + ) + + def test_show_port_pair(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pc: + req = self.new_show_request( + 'port_pairs', pc['port_pair']['id'] + ) + res = self.deserialize( + self.fmt, req.get_response(self.ext_api) + ) + for k, v in six.iteritems(pc['port_pair']): + self.assertEqual(res['port_pair'][k], v) + + def test_show_port_pair_noexist(self): + req = self.new_show_request( + 'port_pairs', '1' + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 404) + + def test_update_port_pair(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'name': 'test1', + 'description': 'desc1', + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pc: + updates = { + 'name': 'test2', + 'description': 'desc2' + } + req = self.new_update_request( + 'port_pairs', {'port_pair': updates}, + pc['port_pair']['id'] + ) + res = self.deserialize( + self.fmt, + req.get_response(self.ext_api) + ) + expected = pc['port_pair'] + expected.update(updates) + for k, v in six.iteritems(expected): + self.assertEqual(res['port_pair'][k], v) + req = self.new_show_request( + 'port_pairs', pc['port_pair']['id'] + ) + res = self.deserialize( + self.fmt, req.get_response(self.ext_api) + ) + for k, v in six.iteritems(expected): + self.assertEqual(res['port_pair'][k], v) + + def test_update_port_pair_service_function_parameters(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'name': 'test1', + 'description': 'desc1', + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pc: + updates = { + 'service_function_parameters': { + 'correlation': None + } + } + req = self.new_update_request( + 'port_pairs', {'port_pair': updates}, + pc['port_pair']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 400) + + def test_update_port_pair_ingress(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'name': 'test1', + 'description': 'desc1', + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pc: + updates = { + 'ingress': dst_port['port']['id'] + } + req = self.new_update_request( + 'port_pairs', {'port_pair': updates}, + pc['port_pair']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 400) + + def test_update_port_pair_egress(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'name': 'test1', + 'description': 'desc1', + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pc: + updates = { + 'egress': src_port['port']['id'] + } + req = self.new_update_request( + 'port_pairs', {'port_pair': updates}, + pc['port_pair']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 400) + + def test_delete_port_pair(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }, do_delete=False) as pc: + req = self.new_delete_request( + 'port_pairs', pc['port_pair']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 204) + req = self.new_show_request( + 'port_pairs', pc['port_pair']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 404) + + def test_delete_port_pair_noexist(self): + req = self.new_delete_request( + 'port_pairs', '1' + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 404) + + def test_delete_port_pair_port_pair_group_exist(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }) as pp: + with self.port_pair_group(port_pair_group={ + 'port_pairs': [pp['port_pair']['id']] + }): + req = self.new_delete_request( + 'port_pairs', pp['port_pair']['id'] + ) + res = req.get_response(self.ext_api) + self.assertEqual(res.status_int, 409) + + def test_delete_ingress_port_pair_exist(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }): + req = self.new_delete_request( + 'ports', src_port['port']['id'] + ) + res = req.get_response(self.api) + self.assertEqual(res.status_int, 500) + + def test_delete_egress_port_pair_exist(self): + with self.port( + name='port1', + device_id='default' + ) as src_port, self.port( + name='port2', + device_id='default' + ) as dst_port: + with self.port_pair(port_pair={ + 'ingress': src_port['port']['id'], + 'egress': dst_port['port']['id'] + }): + req = self.new_delete_request( + 'ports', dst_port['port']['id'] + ) + res = req.get_response(self.api) + self.assertEqual(res.status_int, 500) |