diff options
Diffstat (limited to 'networking_sfc/tests/unit/extensions/test_flowclassifier.py')
| -rw-r--r-- | networking_sfc/tests/unit/extensions/test_flowclassifier.py | 603 |
1 files changed, 0 insertions, 603 deletions
diff --git a/networking_sfc/tests/unit/extensions/test_flowclassifier.py b/networking_sfc/tests/unit/extensions/test_flowclassifier.py deleted file mode 100644 index 7026ac5..0000000 --- a/networking_sfc/tests/unit/extensions/test_flowclassifier.py +++ /dev/null @@ -1,603 +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 copy -import mock -from webob import exc -import webtest - -from oslo_utils import uuidutils - -from neutron.api.v2 import resource as api_res_log -from neutron.common import config -from neutron import manager -from neutron.notifiers import nova as nova_log - -from neutron.tests.unit.api.v2 import test_base as test_api_v2 -from neutron.tests.unit.extensions import base as test_api_v2_extension - -from networking_sfc.extensions import flowclassifier as fc_ext - -_uuid = uuidutils.generate_uuid -_get_path = test_api_v2._get_path - -FLOW_CLASSIFIER_PATH = (fc_ext.FLOW_CLASSIFIER_PREFIX[1:] + '/' + - fc_ext.FLOW_CLASSIFIER_EXT + 's') - - -class FlowClassifierExtensionTestCase( - test_api_v2_extension.ExtensionTestCase -): - fmt = 'json' - - def setUp(self): - self._mock_unncessary_logging() - super(FlowClassifierExtensionTestCase, self).setUp() - self._setUpExtension( - 'networking_sfc.extensions.flowclassifier.' - 'FlowClassifierPluginBase', - fc_ext.FLOW_CLASSIFIER_EXT, - fc_ext.RESOURCE_ATTRIBUTE_MAP, - fc_ext.Flowclassifier, - fc_ext.FLOW_CLASSIFIER_PREFIX[1:], - plural_mappings={} - ) - - def _mock_unncessary_logging(self): - mock_log_cfg_p = mock.patch.object(config, 'LOG') - self.mock_log_cfg = mock_log_cfg_p.start() - - mock_log_manager_p = mock.patch.object(manager, 'LOG') - self.mock_log_manager = mock_log_manager_p.start() - - mock_log_nova_p = mock.patch.object(nova_log, 'LOG') - self.mock_log_nova = mock_log_nova_p.start() - - mock_log_api_res_log_p = mock.patch.object(api_res_log, 'LOG') - self.mock_log_api_res_log = mock_log_api_res_log_p.start() - - def _get_expected_flow_classifier(self, data): - source_port_range_min = data['flow_classifier'].get( - 'source_port_range_min') - if source_port_range_min is not None: - source_port_range_min = int(source_port_range_min) - source_port_range_max = data['flow_classifier'].get( - 'source_port_range_max') - if source_port_range_max is not None: - source_port_range_max = int(source_port_range_max) - destination_port_range_min = data['flow_classifier'].get( - 'destination_port_range_min') - if destination_port_range_min is not None: - destination_port_range_min = int(destination_port_range_min) - destination_port_range_max = data['flow_classifier'].get( - 'destination_port_range_max') - if destination_port_range_max is not None: - destination_port_range_max = int(destination_port_range_max) - - return {'flow_classifier': { - 'name': data['flow_classifier'].get('name') or '', - 'description': data['flow_classifier'].get('description') or '', - 'tenant_id': data['flow_classifier']['tenant_id'], - 'source_port_range_min': source_port_range_min, - 'source_port_range_max': source_port_range_max, - 'destination_port_range_min': destination_port_range_min, - 'destination_port_range_max': destination_port_range_max, - 'l7_parameters': data['flow_classifier'].get( - 'l7_parameters') or {}, - 'destination_ip_prefix': data['flow_classifier'].get( - 'destination_ip_prefix'), - 'source_ip_prefix': data['flow_classifier'].get( - 'source_ip_prefix'), - 'logical_source_port': data['flow_classifier'].get( - 'logical_source_port'), - 'logical_destination_port': data['flow_classifier'].get( - 'logical_destination_port'), - 'ethertype': data['flow_classifier'].get( - 'ethertype') or 'IPv4', - 'protocol': data['flow_classifier'].get( - 'protocol') - }} - - def _clean_expected_flow_classifier(self, expected_flow_classifier): - if 'logical_source_port' in expected_flow_classifier: - del expected_flow_classifier['logical_source_port'] - if 'logical_destination_port' in expected_flow_classifier: - del expected_flow_classifier['logical_destination_port'] - - def test_create_flow_classifier(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'tenant_id': _uuid(), - }} - expected_data = self._get_expected_flow_classifier(data) - return_value = copy.copy(expected_data['flow_classifier']) - return_value.update({'id': flowclassifier_id}) - self._clean_expected_flow_classifier(return_value) - instance = self.plugin.return_value - instance.create_flow_classifier.return_value = return_value - res = self.api.post( - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - instance.create_flow_classifier.assert_called_with( - mock.ANY, - flow_classifier=expected_data) - self.assertEqual(res.status_int, exc.HTTPCreated.code) - res = self.deserialize(res) - self.assertIn('flow_classifier', res) - self.assertEqual(return_value, res['flow_classifier']) - - def test_create_flow_classifier_port_string(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'source_port_range_min': '100', - 'source_port_range_max': '200', - 'destination_port_range_min': '100', - 'destination_port_range_max': '200', - 'tenant_id': _uuid(), - }} - expected_data = self._get_expected_flow_classifier(data) - return_value = copy.copy(expected_data['flow_classifier']) - return_value.update({'id': flowclassifier_id}) - self._clean_expected_flow_classifier(return_value) - instance = self.plugin.return_value - instance.create_flow_classifier.return_value = return_value - res = self.api.post( - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - instance.create_flow_classifier.assert_called_with( - mock.ANY, - flow_classifier=expected_data) - self.assertEqual(res.status_int, exc.HTTPCreated.code) - res = self.deserialize(res) - self.assertIn('flow_classifier', res) - self.assertEqual(return_value, res['flow_classifier']) - - def test_create_flow_classifier_ip_prefix_with_mask(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'source_ip_prefix': '10.0.0.0/8', - 'tenant_id': _uuid(), - }} - expected_data = self._get_expected_flow_classifier(data) - return_value = copy.copy(expected_data['flow_classifier']) - return_value.update({'id': flowclassifier_id}) - self._clean_expected_flow_classifier(return_value) - instance = self.plugin.return_value - instance.create_flow_classifier.return_value = return_value - res = self.api.post( - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - instance.create_flow_classifier.assert_called_with( - mock.ANY, - flow_classifier=expected_data) - self.assertEqual(res.status_int, exc.HTTPCreated.code) - res = self.deserialize(res) - self.assertIn('flow_classifier', res) - self.assertEqual(return_value, res['flow_classifier']) - - def test_create_flow_classifier_non_l7_parameters(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'tenant_id': _uuid(), - 'l7_parameters': None - }} - expected_data = self._get_expected_flow_classifier(data) - return_value = copy.copy(expected_data['flow_classifier']) - return_value.update({'id': flowclassifier_id}) - self._clean_expected_flow_classifier(return_value) - instance = self.plugin.return_value - instance.create_flow_classifier.return_value = return_value - res = self.api.post( - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - instance.create_flow_classifier.assert_called_with( - mock.ANY, - flow_classifier=expected_data) - self.assertEqual(res.status_int, exc.HTTPCreated.code) - res = self.deserialize(res) - self.assertIn('flow_classifier', res) - self.assertEqual(return_value, res['flow_classifier']) - - def test_create_flow_classifier_default_ethertype(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'tenant_id': _uuid(), - 'ethertype': 'IPv4' - }} - expected_data = self._get_expected_flow_classifier(data) - return_value = copy.copy(expected_data['flow_classifier']) - return_value.update({'id': flowclassifier_id}) - self._clean_expected_flow_classifier(return_value) - instance = self.plugin.return_value - instance.create_flow_classifier.return_value = return_value - res = self.api.post( - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - instance.create_flow_classifier.assert_called_with( - mock.ANY, - flow_classifier=expected_data) - self.assertEqual(res.status_int, exc.HTTPCreated.code) - res = self.deserialize(res) - self.assertIn('flow_classifier', res) - self.assertEqual(return_value, res['flow_classifier']) - - def test_create_flow_classifier_all_fields(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'name': 'test1', - 'description': 'desc', - 'tenant_id': _uuid(), - 'source_port_range_min': 100, - 'source_port_range_max': 200, - 'destination_port_range_min': 100, - 'destination_port_range_max': 200, - 'l7_parameters': {}, - 'destination_ip_prefix': '10.0.0.0/8', - 'source_ip_prefix': '10.0.0.0/8', - 'logical_source_port': _uuid(), - 'logical_destination_port': _uuid(), - 'ethertype': None, - 'protocol': None - }} - expected_data = self._get_expected_flow_classifier(data) - return_value = copy.copy(expected_data['flow_classifier']) - return_value.update({'id': flowclassifier_id}) - self._clean_expected_flow_classifier(return_value) - instance = self.plugin.return_value - instance.create_flow_classifier.return_value = return_value - res = self.api.post( - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - instance.create_flow_classifier.assert_called_with( - mock.ANY, - flow_classifier=expected_data) - self.assertEqual(res.status_int, exc.HTTPCreated.code) - res = self.deserialize(res) - self.assertIn('flow_classifier', res) - self.assertEqual(return_value, res['flow_classifier']) - - def test_create_flow_classifier_invalid_l7_parameters(self): - data = {'flow_classifier': { - 'l7_parameters': {'abc': 'def'}, - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.post, - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_create_flow_classifier_invalid_protocol(self): - data = {'flow_classifier': { - 'protocol': 'unknown', - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.post, - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_create_flow_classifier_invalid_ethertype(self): - data = {'flow_classifier': { - 'ethertype': 'unknown', - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.post, - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_create_flow_classifier_port_small(self): - data = {'flow_classifier': { - 'source_port_range_min': -1, - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.post, - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_create_flow_classifier_port_large(self): - data = {'flow_classifier': { - 'source_port_range_min': 65536, - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.post, - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_create_flow_classifier_ip_prefix_no_cidr(self): - data = {'flow_classifier': { - 'source_ip_prefix': '10.0.0.0', - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.post, - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_create_flow_classifier_ip_prefix_invalid_cidr(self): - data = {'flow_classifier': { - 'source_ip_prefix': '10.0.0.0/33', - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.post, - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_create_flow_classifier_port_id_nouuid(self): - data = {'flow_classifier': { - 'logical_source_port': 'unknown', - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.post, - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_list(self): - flowclassifier_id = _uuid() - return_value = [{ - 'tenant_id': _uuid(), - 'id': flowclassifier_id - }] - instance = self.plugin.return_value - instance.get_flow_classifiers.return_value = return_value - - res = self.api.get( - _get_path(FLOW_CLASSIFIER_PATH, fmt=self.fmt)) - - instance.get_flow_classifiers.assert_called_with( - mock.ANY, - fields=mock.ANY, - filters=mock.ANY - ) - self.assertEqual(res.status_int, exc.HTTPOk.code) - res = self.deserialize(res) - self.assertIn('flow_classifiers', res) - self.assertEqual(res['flow_classifiers'], return_value) - - def test_flow_classifier_get(self): - flowclassifier_id = _uuid() - return_value = { - 'tenant_id': _uuid(), - 'id': flowclassifier_id - } - - instance = self.plugin.return_value - instance.get_flow_classifier.return_value = return_value - - res = self.api.get( - _get_path( - FLOW_CLASSIFIER_PATH, - id=flowclassifier_id, fmt=self.fmt - ) - ) - - instance.get_flow_classifier.assert_called_with( - mock.ANY, - flowclassifier_id, - fields=mock.ANY - ) - self.assertEqual(res.status_int, exc.HTTPOk.code) - res = self.deserialize(res) - self.assertIn('flow_classifier', res) - self.assertEqual(return_value, res['flow_classifier']) - - def test_flow_classifier_update(self): - flowclassifier_id = _uuid() - update_data = {'flow_classifier': { - 'name': 'new_name', - 'description': 'new_desc', - }} - return_value = { - 'tenant_id': _uuid(), - 'id': flowclassifier_id - } - - instance = self.plugin.return_value - instance.update_flow_classifier.return_value = return_value - - res = self.api.put( - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(update_data)) - - instance.update_flow_classifier.assert_called_with( - mock.ANY, flowclassifier_id, - flow_classifier=update_data) - self.assertEqual(res.status_int, exc.HTTPOk.code) - res = self.deserialize(res) - self.assertIn('flow_classifier', res) - self.assertEqual(res['flow_classifier'], return_value) - - def test_flow_classifier_update_source_port_range_min(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'source_port_range_min': 100, - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_update_source_port_range_max(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'source_port_range_max': 100, - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_update_destination_port_range_min(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'destination_port_range_min': 100, - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_update_destination_port_range_max(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'destination_port_range_max': 100, - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_update_source_ip_prefix(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'source_ip_prefix': '10.0.0.0/8', - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_update_destination_ip_prefix(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'destination_ip_prefix': '10.0.0.0/8', - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_update_logical_source_port(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'logical_source_port': _uuid(), - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_update_logical_destination_port(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'logical_destination_port': _uuid(), - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_update_ethertype(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'ethertype': None, - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_update_protocol(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'protococol': None, - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_update_l7_parameters(self): - flowclassifier_id = _uuid() - data = {'flow_classifier': { - 'l7_parameters': {}, - 'tenant_id': _uuid() - }} - self.assertRaises( - webtest.app.AppError, - self.api.put, - _get_path(FLOW_CLASSIFIER_PATH, id=flowclassifier_id, - fmt=self.fmt), - self.serialize(data), - content_type='application/%s' % self.fmt) - - def test_flow_classifier_delete(self): - self._test_entity_delete('flow_classifier') |