diff options
Diffstat (limited to 'networking_sfc/tests/unit/services/flowclassifier')
3 files changed, 0 insertions, 326 deletions
diff --git a/networking_sfc/tests/unit/services/flowclassifier/__init__.py b/networking_sfc/tests/unit/services/flowclassifier/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/networking_sfc/tests/unit/services/flowclassifier/__init__.py +++ /dev/null diff --git a/networking_sfc/tests/unit/services/flowclassifier/test_driver_manager.py b/networking_sfc/tests/unit/services/flowclassifier/test_driver_manager.py deleted file mode 100644 index 56dd991..0000000 --- a/networking_sfc/tests/unit/services/flowclassifier/test_driver_manager.py +++ /dev/null @@ -1,158 +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 pkg_resources -import six -import stevedore - -from oslo_config import cfg - -from neutron.tests import base - -from networking_sfc.services.flowclassifier.common import config as fc_config -from networking_sfc.services.flowclassifier.common import exceptions as fc_exc -from networking_sfc.services.flowclassifier import driver_manager as fc_driver - - -class DriverManagerTestCase(base.BaseTestCase): - def setUp(self): - super(DriverManagerTestCase, self).setUp() - - @contextlib.contextmanager - def driver_manager_context(self, drivers): - cfg.CONF.register_opts(fc_config.FLOWCLASSIFIER_DRIVER_OPTS, - 'flowclassifier') - backup_driver_names = cfg.CONF.flowclassifier.drivers - driver_names = [ - driver_name for driver_name in six.iterkeys(drivers) - ] - cfg.CONF.set_override('drivers', driver_names, 'flowclassifier') - iter_entry_points = pkg_resources.iter_entry_points - find_entry_points = stevedore.ExtensionManager._find_entry_points - pkg_resources.iter_entry_points = mock.Mock() - stevedore.ExtensionManager._find_entry_points = mock.Mock() - driver_entry_points = [] - for driver_name in driver_names: - driver_class = mock.Mock() - ep = mock.Mock() - ep.name = driver_name - ep.resolve.return_value = driver_class - driver_class.return_value = drivers[driver_name] - drivers[driver_name].native_bulk_support = True - driver_entry_points.append(ep) - pkg_resources.iter_entry_points.return_value = driver_entry_points - stevedore.ExtensionManager._find_entry_points.return_value = ( - driver_entry_points - ) - yield fc_driver.FlowClassifierDriverManager() - cfg.CONF.set_override('drivers', backup_driver_names, 'flowclassifier') - pkg_resources.iter_entry_points = iter_entry_points - stevedore.ExtensionManager._find_entry_points = find_entry_points - - def test_initialize_called(self): - mock_driver1 = mock.Mock() - mock_driver2 = mock.Mock() - with self.driver_manager_context({ - 'dummy1': mock_driver1, - 'dummy2': mock_driver2 - }) as manager: - manager.initialize() - mock_driver1.initialize.assert_called_once_with() - mock_driver2.initialize.assert_called_once_with() - - def test_create_flow_classifier_called(self): - mock_driver1 = mock.Mock() - mock_driver2 = mock.Mock() - with self.driver_manager_context({ - 'dummy1': mock_driver1, - 'dummy2': mock_driver2 - }) as manager: - mocked_context = mock.Mock() - manager.create_flow_classifier(mocked_context) - mock_driver1.create_flow_classifier.assert_called_once_with( - mocked_context) - mock_driver2.create_flow_classifier.assert_called_once_with( - mocked_context) - - def test_create_flow_classifier_exception(self): - mock_driver = mock.Mock() - mock_driver.create_flow_classifier = mock.Mock( - side_effect=fc_exc.FlowClassifierException - ) - with self.driver_manager_context({ - 'dummy': mock_driver, - }) as manager: - mocked_context = mock.Mock() - self.assertRaises( - fc_exc.FlowClassifierDriverError, - manager.create_flow_classifier, mocked_context - ) - - def test_update_flow_classifier_called(self): - mock_driver1 = mock.Mock() - mock_driver2 = mock.Mock() - with self.driver_manager_context({ - 'dummy1': mock_driver1, - 'dummy2': mock_driver2 - }) as manager: - mocked_context = mock.Mock() - manager.update_flow_classifier(mocked_context) - mock_driver1.update_flow_classifier.assert_called_once_with( - mocked_context) - mock_driver2.update_flow_classifier.assert_called_once_with( - mocked_context) - - def test_update_flow_classifier_exception(self): - mock_driver = mock.Mock() - mock_driver.update_flow_classifier = mock.Mock( - side_effect=fc_exc.FlowClassifierException - ) - with self.driver_manager_context({ - 'dummy': mock_driver, - }) as manager: - mocked_context = mock.Mock() - self.assertRaises( - fc_exc.FlowClassifierDriverError, - manager.update_flow_classifier, mocked_context - ) - - def test_delete_flow_classifier_called(self): - mock_driver1 = mock.Mock() - mock_driver2 = mock.Mock() - with self.driver_manager_context({ - 'dummy1': mock_driver1, - 'dummy2': mock_driver2 - }) as manager: - mocked_context = mock.Mock() - manager.delete_flow_classifier(mocked_context) - mock_driver1.delete_flow_classifier.assert_called_once_with( - mocked_context) - mock_driver2.delete_flow_classifier.assert_called_once_with( - mocked_context) - - def test_delete_flow_classifier_exception(self): - mock_driver = mock.Mock() - mock_driver.delete_flow_classifier = mock.Mock( - side_effect=fc_exc.FlowClassifierException - ) - with self.driver_manager_context({ - 'dummy': mock_driver, - }) as manager: - mocked_context = mock.Mock() - self.assertRaises( - fc_exc.FlowClassifierDriverError, - manager.delete_flow_classifier, mocked_context - ) diff --git a/networking_sfc/tests/unit/services/flowclassifier/test_plugin.py b/networking_sfc/tests/unit/services/flowclassifier/test_plugin.py deleted file mode 100644 index 2c814f8..0000000 --- a/networking_sfc/tests/unit/services/flowclassifier/test_plugin.py +++ /dev/null @@ -1,168 +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 networking_sfc.services.flowclassifier.common import context as fc_ctx -from networking_sfc.services.flowclassifier.common import exceptions as fc_exc -from networking_sfc.tests.unit.db import test_flowclassifier_db - -FLOWCLASSIFIER_PLUGIN_KLASS = ( - "networking_sfc.services.flowclassifier." - "plugin.FlowClassifierPlugin" -) - - -class FlowClassifierPluginTestCase( - test_flowclassifier_db.FlowClassifierDbPluginTestCase -): - def setUp( - self, core_plugin=None, flowclassifier_plugin=None, ext_mgr=None - ): - if not flowclassifier_plugin: - flowclassifier_plugin = FLOWCLASSIFIER_PLUGIN_KLASS - self.driver_manager_p = mock.patch( - 'networking_sfc.services.flowclassifier.driver_manager.' - 'FlowClassifierDriverManager' - ) - self.fake_driver_manager_class = self.driver_manager_p.start() - self.fake_driver_manager = mock.Mock() - self.fake_driver_manager_class.return_value = self.fake_driver_manager - self.plugin_context = None - super(FlowClassifierPluginTestCase, self).setUp( - core_plugin=core_plugin, - flowclassifier_plugin=flowclassifier_plugin, - ext_mgr=ext_mgr - ) - - def _record_context(self, plugin_context): - self.plugin_context = plugin_context - - def test_create_flow_classifier_driver_manager_called(self): - self.fake_driver_manager.create_flow_classifier = mock.Mock( - side_effect=self._record_context) - with self.flow_classifier(flow_classifier={}) as fc: - driver_manager = self.fake_driver_manager - driver_manager.create_flow_classifier.assert_called_once_with( - mock.ANY - ) - self.assertIsInstance( - self.plugin_context, fc_ctx.FlowClassifierContext - ) - self.assertIn('flow_classifier', fc) - self.assertEqual( - self.plugin_context.current, fc['flow_classifier']) - - def test_create_flow_classifier_driver_manager_exception(self): - self.fake_driver_manager.create_flow_classifier = mock.Mock( - side_effect=fc_exc.FlowClassifierDriverError( - method='create_flow_classifier' - ) - ) - self._create_flow_classifier( - self.fmt, {}, expected_res_status=500) - self._test_list_resources('flow_classifier', []) - driver_manager = self.fake_driver_manager - driver_manager.delete_flow_classifier.assert_called_once_with( - mock.ANY - ) - - def test_update_flow_classifier_driver_manager_called(self): - self.fake_driver_manager.update_flow_classifier = mock.Mock( - side_effect=self._record_context) - with self.flow_classifier(flow_classifier={'name': 'test1'}) as fc: - req = self.new_update_request( - 'flow_classifiers', {'flow_classifier': {'name': 'test2'}}, - fc['flow_classifier']['id'] - ) - res = self.deserialize( - self.fmt, - req.get_response(self.ext_api) - ) - driver_manager = self.fake_driver_manager - driver_manager.update_flow_classifier.assert_called_once_with( - mock.ANY - ) - self.assertIsInstance( - self.plugin_context, fc_ctx.FlowClassifierContext - ) - self.assertIn('flow_classifier', fc) - self.assertIn('flow_classifier', res) - self.assertEqual( - self.plugin_context.current, res['flow_classifier']) - self.assertEqual( - self.plugin_context.original, fc['flow_classifier']) - - def test_update_flow_classifier_driver_manager_exception(self): - self.fake_driver_manager.update_flow_classifier = mock.Mock( - side_effect=fc_exc.FlowClassifierDriverError( - method='update_flow_classifier' - ) - ) - with self.flow_classifier(flow_classifier={ - 'name': 'test1' - }) as fc: - self.assertIn('flow_classifier', fc) - original_flow_classifier = fc['flow_classifier'] - req = self.new_update_request( - 'flow_classifiers', {'flow_classifier': {'name': 'test2'}}, - fc['flow_classifier']['id'] - ) - updated_flow_classifier = copy.copy(original_flow_classifier) - updated_flow_classifier['name'] = 'test2' - res = req.get_response(self.ext_api) - self.assertEqual(res.status_int, 500) - res = self._list('flow_classifiers') - self.assertIn('flow_classifiers', res) - self.assertItemsEqual( - res['flow_classifiers'], [updated_flow_classifier]) - - def test_delete_flow_classifer_driver_manager_called(self): - self.fake_driver_manager.delete_flow_classifier = mock.Mock( - side_effect=self._record_context) - with self.flow_classifier( - flow_classifier={}, do_delete=False - ) as fc: - req = self.new_delete_request( - 'flow_classifiers', fc['flow_classifier']['id'] - ) - res = req.get_response(self.ext_api) - self.assertEqual(res.status_int, 204) - driver_manager = self.fake_driver_manager - driver_manager.delete_flow_classifier.assert_called_once_with( - mock.ANY - ) - self.assertIsInstance( - self.plugin_context, fc_ctx.FlowClassifierContext - ) - self.assertIn('flow_classifier', fc) - self.assertEqual( - self.plugin_context.current, fc['flow_classifier']) - - def test_delete_flow_classifier_driver_manager_exception(self): - self.fake_driver_manager.delete_flow_classifier = mock.Mock( - side_effect=fc_exc.FlowClassifierDriverError( - method='delete_flow_classifier' - ) - ) - with self.flow_classifier(flow_classifier={ - 'name': 'test1' - }, do_delete=False) as fc: - req = self.new_delete_request( - 'flow_classifiers', fc['flow_classifier']['id'] - ) - res = req.get_response(self.ext_api) - self.assertEqual(res.status_int, 500) - self._test_list_resources('flow_classifier', [fc]) |