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, 0 insertions, 1490 deletions
diff --git a/networking_sfc/tests/unit/db/test_sfc_db.py b/networking_sfc/tests/unit/db/test_sfc_db.py deleted file mode 100644 index 70d57e3..0000000 --- a/networking_sfc/tests/unit/db/test_sfc_db.py +++ /dev/null @@ -1,1490 +0,0 @@ -# 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) |