aboutsummaryrefslogtreecommitdiffstats
path: root/networking_sfc/tests/unit/extensions/test_flowclassifier.py
diff options
context:
space:
mode:
Diffstat (limited to 'networking_sfc/tests/unit/extensions/test_flowclassifier.py')
-rw-r--r--networking_sfc/tests/unit/extensions/test_flowclassifier.py603
1 files changed, 603 insertions, 0 deletions
diff --git a/networking_sfc/tests/unit/extensions/test_flowclassifier.py b/networking_sfc/tests/unit/extensions/test_flowclassifier.py
new file mode 100644
index 0000000..7026ac5
--- /dev/null
+++ b/networking_sfc/tests/unit/extensions/test_flowclassifier.py
@@ -0,0 +1,603 @@
+# 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')