# 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. from unittest.mock import Mock from unittest.mock import sentinel from stevedore.dispatch import DispatchExtensionManager from stevedore.dispatch import NameDispatchExtensionManager from stevedore.extension import Extension from stevedore.tests import utils from stevedore import DriverManager from stevedore import EnabledExtensionManager from stevedore import ExtensionManager from stevedore import HookManager from stevedore import NamedExtensionManager test_extension = Extension('test_extension', None, None, None) test_extension2 = Extension('another_one', None, None, None) mock_entry_point = Mock(module_name='test.extension', attrs=['obj']) a_driver = Extension('test_driver', mock_entry_point, sentinel.driver_plugin, sentinel.driver_obj) # base ExtensionManager class TestTestManager(utils.TestCase): def test_instance_should_use_supplied_extensions(self): extensions = [test_extension, test_extension2] em = ExtensionManager.make_test_instance(extensions) self.assertEqual(extensions, em.extensions) def test_instance_should_have_default_namespace(self): em = ExtensionManager.make_test_instance([]) self.assertEqual(em.namespace, 'TESTING') def test_instance_should_use_supplied_namespace(self): namespace = 'testing.1.2.3' em = ExtensionManager.make_test_instance([], namespace=namespace) self.assertEqual(namespace, em.namespace) def test_extension_name_should_be_listed(self): em = ExtensionManager.make_test_instance([test_extension]) self.assertIn(test_extension.name, em.names()) def test_iterator_should_yield_extension(self): em = ExtensionManager.make_test_instance([test_extension]) self.assertEqual(test_extension, next(iter(em))) def test_manager_should_allow_name_access(self): em = ExtensionManager.make_test_instance([test_extension]) self.assertEqual(test_extension, em[test_extension.name]) def test_manager_should_call(self): em = ExtensionManager.make_test_instance([test_extension]) func = Mock() em.map(func) func.assert_called_once_with(test_extension) def test_manager_should_call_all(self): em = ExtensionManager.make_test_instance([test_extension2, test_extension]) func = Mock() em.map(func) func.assert_any_call(test_extension2) func.assert_any_call(test_extension) def test_manager_return_values(self): def mapped(ext, *args, **kwds): return ext.name em = ExtensionManager.make_test_instance([test_extension2, test_extension]) results = em.map(mapped) self.assertEqual(sorted(results), ['another_one', 'test_extension']) def test_manager_should_eat_exceptions(self): em = ExtensionManager.make_test_instance([test_extension]) func = Mock(side_effect=RuntimeError('hard coded error')) results = em.map(func, 1, 2, a='A', b='B') self.assertEqual(results, []) def test_manager_should_propagate_exceptions(self): em = ExtensionManager.make_test_instance([test_extension], propagate_map_exceptions=True) self.skipTest('Skipping temporarily') func = Mock(side_effect=RuntimeError('hard coded error')) em.map(func, 1, 2, a='A', b='B') # NamedExtensionManager def test_named_manager_should_use_supplied_extensions(self): extensions = [test_extension, test_extension2] em = NamedExtensionManager.make_test_instance(extensions) self.assertEqual(extensions, em.extensions) def test_named_manager_should_have_default_namespace(self): em = NamedExtensionManager.make_test_instance([]) self.assertEqual(em.namespace, 'TESTING') def test_named_manager_should_use_supplied_namespace(self): namespace = 'testing.1.2.3' em = NamedExtensionManager.make_test_instance([], namespace=namespace) self.assertEqual(namespace, em.namespace) def test_named_manager_should_populate_names(self): extensions = [test_extension, test_extension2] em = NamedExtensionManager.make_test_instance(extensions) self.assertEqual(em.names(), ['test_extension', 'another_one']) # HookManager def test_hook_manager_should_use_supplied_extensions(self): extensions = [test_extension, test_extension2] em = HookManager.make_test_instance(extensions) self.assertEqual(extensions, em.extensions) def test_hook_manager_should_be_first_extension_name(self): extensions = [test_extension, test_extension2] em = HookManager.make_test_instance(extensions) # This will raise KeyError if the names don't match assert(em[test_extension.name]) def test_hook_manager_should_have_default_namespace(self): em = HookManager.make_test_instance([test_extension]) self.assertEqual(em.namespace, 'TESTING') def test_hook_manager_should_use_supplied_namespace(self): namespace = 'testing.1.2.3' em = HookManager.make_test_instance([test_extension], namespace=namespace) self.assertEqual(namespace, em.namespace) def test_hook_manager_should_return_named_extensions(self): hook1 = Extension('captain', None, None, None) hook2 = Extension('captain', None, None, None) em = HookManager.make_test_instance([hook1, hook2]) self.assertEqual([hook1, hook2], em['captain']) # DriverManager def test_driver_manager_should_use_supplied_extension(self): em = DriverManager.make_test_instance(a_driver) self.assertEqual([a_driver], em.extensions) def test_driver_manager_should_have_default_namespace(self): em = DriverManager.make_test_instance(a_driver) self.assertEqual(em.namespace, 'TESTING') def test_driver_manager_should_use_supplied_namespace(self): namespace = 'testing.1.2.3' em = DriverManager.make_test_instance(a_driver, namespace=namespace) self.assertEqual(namespace, em.namespace) def test_instance_should_use_driver_name(self): em = DriverManager.make_test_instance(a_driver) self.assertEqual(['test_driver'], em.names()) def test_instance_call(self): def invoke(ext, *args, **kwds): return ext.name, args, kwds em = DriverManager.make_test_instance(a_driver) result = em(invoke, 'a', b='C') self.assertEqual(result, ('test_driver', ('a',), {'b': 'C'})) def test_instance_driver_property(self): em = DriverManager.make_test_instance(a_driver) self.assertEqual(sentinel.driver_obj, em.driver) # EnabledExtensionManager def test_enabled_instance_should_use_supplied_extensions(self): extensions = [test_extension, test_extension2] em = EnabledExtensionManager.make_test_instance(extensions) self.assertEqual(extensions, em.extensions) # DispatchExtensionManager def test_dispatch_instance_should_use_supplied_extensions(self): extensions = [test_extension, test_extension2] em = DispatchExtensionManager.make_test_instance(extensions) self.assertEqual(extensions, em.extensions) def test_dispatch_map_should_invoke_filter_for_extensions(self): em = DispatchExtensionManager.make_test_instance([test_extension, test_extension2]) filter_func = Mock(return_value=False) args = ('A',) kw = {'big': 'Cheese'} em.map(filter_func, None, *args, **kw) filter_func.assert_any_call(test_extension, *args, **kw) filter_func.assert_any_call(test_extension2, *args, **kw) # NameDispatchExtensionManager def test_name_dispatch_instance_should_use_supplied_extensions(self): extensions = [test_extension, test_extension2] em = NameDispatchExtensionManager.make_test_instance(extensions) self.assertEqual(extensions, em.extensions) def test_name_dispatch_instance_should_build_extension_name_map(self): extensions = [test_extension, test_extension2] em = NameDispatchExtensionManager.make_test_instance(extensions) self.assertEqual(test_extension, em.by_name[test_extension.name]) self.assertEqual(test_extension2, em.by_name[test_extension2.name]) def test_named_dispatch_map_should_invoke_filter_for_extensions(self): em = NameDispatchExtensionManager.make_test_instance([test_extension, test_extension2]) func = Mock() args = ('A',) kw = {'BIGGER': 'Cheese'} em.map(['test_extension'], func, *args, **kw) func.assert_called_once_with(test_extension, *args, **kw)