290 lines
10 KiB
Python
290 lines
10 KiB
Python
# 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.
|
|
|
|
"""Tests for stevedore.extension
|
|
"""
|
|
|
|
import operator
|
|
from unittest import mock
|
|
|
|
try:
|
|
# For python 3.8 and later
|
|
import importlib.metadata as importlib_metadata
|
|
except ImportError:
|
|
# For everyone else
|
|
import importlib_metadata
|
|
|
|
from stevedore import exception
|
|
from stevedore import extension
|
|
from stevedore.tests import utils
|
|
|
|
|
|
ALL_NAMES = ['e1', 't1', 't2']
|
|
WORKING_NAMES = ['t1', 't2']
|
|
|
|
|
|
class FauxExtension(object):
|
|
def __init__(self, *args, **kwds):
|
|
self.args = args
|
|
self.kwds = kwds
|
|
|
|
def get_args_and_data(self, data):
|
|
return self.args, self.kwds, data
|
|
|
|
|
|
class BrokenExtension(object):
|
|
def __init__(self, *args, **kwds):
|
|
raise IOError("Did not create")
|
|
|
|
|
|
class TestCallback(utils.TestCase):
|
|
def test_detect_plugins(self):
|
|
em = extension.ExtensionManager('stevedore.test.extension')
|
|
names = sorted(em.names())
|
|
self.assertEqual(names, ALL_NAMES)
|
|
|
|
def test_get_by_name(self):
|
|
em = extension.ExtensionManager('stevedore.test.extension')
|
|
e = em['t1']
|
|
self.assertEqual(e.name, 't1')
|
|
|
|
def test_list_entry_points(self):
|
|
em = extension.ExtensionManager('stevedore.test.extension')
|
|
n = em.list_entry_points()
|
|
self.assertEqual(set(['e1', 'e2', 't1', 't2']),
|
|
set(map(operator.attrgetter("name"), n)))
|
|
self.assertEqual(4, len(n))
|
|
|
|
def test_list_entry_points_names(self):
|
|
em = extension.ExtensionManager('stevedore.test.extension')
|
|
names = em.entry_points_names()
|
|
self.assertEqual(set(['e1', 'e2', 't1', 't2']), set(names))
|
|
self.assertEqual(4, len(names))
|
|
|
|
def test_contains_by_name(self):
|
|
em = extension.ExtensionManager('stevedore.test.extension')
|
|
self.assertEqual('t1' in em, True)
|
|
|
|
def test_get_by_name_missing(self):
|
|
em = extension.ExtensionManager('stevedore.test.extension')
|
|
try:
|
|
em['t3']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
assert False, 'Failed to raise KeyError'
|
|
|
|
def test_load_multiple_times_entry_points(self):
|
|
# We expect to get the same EntryPoint object because we save them
|
|
# in the cache.
|
|
em1 = extension.ExtensionManager('stevedore.test.extension')
|
|
eps1 = [ext.entry_point for ext in em1]
|
|
em2 = extension.ExtensionManager('stevedore.test.extension')
|
|
eps2 = [ext.entry_point for ext in em2]
|
|
self.assertIs(eps1[0], eps2[0])
|
|
|
|
def test_load_multiple_times_plugins(self):
|
|
# We expect to get the same plugin object (module or class)
|
|
# because the underlying import machinery will cache the values.
|
|
em1 = extension.ExtensionManager('stevedore.test.extension')
|
|
plugins1 = [ext.plugin for ext in em1]
|
|
em2 = extension.ExtensionManager('stevedore.test.extension')
|
|
plugins2 = [ext.plugin for ext in em2]
|
|
self.assertIs(plugins1[0], plugins2[0])
|
|
|
|
def test_use_cache(self):
|
|
# If we insert something into the cache of entry points,
|
|
# the manager should not have to call into entrypoints
|
|
# to find the plugins.
|
|
cache = extension.ExtensionManager.ENTRY_POINT_CACHE
|
|
cache['stevedore.test.faux'] = []
|
|
with mock.patch('stevedore._cache.get_group_all',
|
|
side_effect=
|
|
AssertionError('called get_group_all')):
|
|
em = extension.ExtensionManager('stevedore.test.faux')
|
|
names = em.names()
|
|
self.assertEqual(names, [])
|
|
|
|
def test_iterable(self):
|
|
em = extension.ExtensionManager('stevedore.test.extension')
|
|
names = sorted(e.name for e in em)
|
|
self.assertEqual(names, ALL_NAMES)
|
|
|
|
def test_invoke_on_load(self):
|
|
em = extension.ExtensionManager('stevedore.test.extension',
|
|
invoke_on_load=True,
|
|
invoke_args=('a',),
|
|
invoke_kwds={'b': 'B'},
|
|
)
|
|
self.assertEqual(len(em.extensions), 2)
|
|
for e in em.extensions:
|
|
self.assertEqual(e.obj.args, ('a',))
|
|
self.assertEqual(e.obj.kwds, {'b': 'B'})
|
|
|
|
def test_map_return_values(self):
|
|
def mapped(ext, *args, **kwds):
|
|
return ext.name
|
|
|
|
em = extension.ExtensionManager('stevedore.test.extension',
|
|
invoke_on_load=True,
|
|
)
|
|
results = em.map(mapped)
|
|
self.assertEqual(sorted(results), WORKING_NAMES)
|
|
|
|
def test_map_arguments(self):
|
|
objs = []
|
|
|
|
def mapped(ext, *args, **kwds):
|
|
objs.append((ext, args, kwds))
|
|
|
|
em = extension.ExtensionManager('stevedore.test.extension',
|
|
invoke_on_load=True,
|
|
)
|
|
em.map(mapped, 1, 2, a='A', b='B')
|
|
self.assertEqual(len(objs), 2)
|
|
names = sorted([o[0].name for o in objs])
|
|
self.assertEqual(names, WORKING_NAMES)
|
|
for o in objs:
|
|
self.assertEqual(o[1], (1, 2))
|
|
self.assertEqual(o[2], {'a': 'A', 'b': 'B'})
|
|
|
|
def test_map_eats_errors(self):
|
|
def mapped(ext, *args, **kwds):
|
|
raise RuntimeError('hard coded error')
|
|
|
|
em = extension.ExtensionManager('stevedore.test.extension',
|
|
invoke_on_load=True,
|
|
)
|
|
results = em.map(mapped, 1, 2, a='A', b='B')
|
|
self.assertEqual(results, [])
|
|
|
|
def test_map_propagate_exceptions(self):
|
|
def mapped(ext, *args, **kwds):
|
|
raise RuntimeError('hard coded error')
|
|
|
|
em = extension.ExtensionManager('stevedore.test.extension',
|
|
invoke_on_load=True,
|
|
propagate_map_exceptions=True
|
|
)
|
|
|
|
try:
|
|
em.map(mapped, 1, 2, a='A', b='B')
|
|
assert False
|
|
except RuntimeError:
|
|
pass
|
|
|
|
def test_map_errors_when_no_plugins(self):
|
|
expected_str = 'No stevedore.test.extension.none extensions found'
|
|
|
|
def mapped(ext, *args, **kwds):
|
|
pass
|
|
|
|
em = extension.ExtensionManager('stevedore.test.extension.none',
|
|
invoke_on_load=True,
|
|
)
|
|
try:
|
|
em.map(mapped, 1, 2, a='A', b='B')
|
|
except exception.NoMatches as err:
|
|
self.assertEqual(expected_str, str(err))
|
|
|
|
def test_map_method(self):
|
|
em = extension.ExtensionManager('stevedore.test.extension',
|
|
invoke_on_load=True,
|
|
)
|
|
|
|
result = em.map_method('get_args_and_data', 42)
|
|
self.assertEqual(set(r[2] for r in result), set([42]))
|
|
|
|
def test_items(self):
|
|
em = extension.ExtensionManager('stevedore.test.extension')
|
|
expected_output = set([(name, em[name]) for name in ALL_NAMES])
|
|
self.assertEqual(expected_output, set(em.items()))
|
|
|
|
|
|
class TestLoadRequirementsNewSetuptools(utils.TestCase):
|
|
# setuptools 11.3 and later
|
|
|
|
def setUp(self):
|
|
super(TestLoadRequirementsNewSetuptools, self).setUp()
|
|
self.mock_ep = mock.Mock(spec=['require', 'resolve', 'load', 'name'])
|
|
self.em = extension.ExtensionManager.make_test_instance([])
|
|
|
|
def test_verify_requirements(self):
|
|
self.em._load_one_plugin(self.mock_ep, False, (), {},
|
|
verify_requirements=True)
|
|
self.mock_ep.require.assert_called_once_with()
|
|
self.mock_ep.resolve.assert_called_once_with()
|
|
|
|
def test_no_verify_requirements(self):
|
|
self.em._load_one_plugin(self.mock_ep, False, (), {},
|
|
verify_requirements=False)
|
|
self.assertEqual(0, self.mock_ep.require.call_count)
|
|
self.mock_ep.resolve.assert_called_once_with()
|
|
|
|
|
|
class TestLoadRequirementsOldSetuptools(utils.TestCase):
|
|
# Before setuptools 11.3
|
|
|
|
def setUp(self):
|
|
super(TestLoadRequirementsOldSetuptools, self).setUp()
|
|
self.mock_ep = mock.Mock(spec=['load', 'name'])
|
|
self.em = extension.ExtensionManager.make_test_instance([])
|
|
|
|
def test_verify_requirements(self):
|
|
self.em._load_one_plugin(self.mock_ep, False, (), {},
|
|
verify_requirements=True)
|
|
self.mock_ep.load.assert_called_once_with()
|
|
|
|
def test_no_verify_requirements(self):
|
|
self.em._load_one_plugin(self.mock_ep, False, (), {},
|
|
verify_requirements=False)
|
|
self.mock_ep.load.assert_called_once_with()
|
|
|
|
|
|
class TestExtensionProperties(utils.TestCase):
|
|
|
|
def setUp(self):
|
|
self.ext1 = extension.Extension(
|
|
'name',
|
|
importlib_metadata.EntryPoint(
|
|
'name', 'module.name:attribute.name [extra]', 'group_name',
|
|
),
|
|
mock.Mock(),
|
|
None,
|
|
)
|
|
self.ext2 = extension.Extension(
|
|
'name',
|
|
importlib_metadata.EntryPoint(
|
|
'name', 'module:attribute', 'group_name',
|
|
),
|
|
mock.Mock(),
|
|
None,
|
|
)
|
|
|
|
def test_module_name(self):
|
|
self.assertEqual('module.name', self.ext1.module_name)
|
|
self.assertEqual('module', self.ext2.module_name)
|
|
|
|
def test_extras(self):
|
|
self.assertEqual(['[extra]'], self.ext1.extras)
|
|
self.assertEqual([], self.ext2.extras)
|
|
|
|
def test_attr(self):
|
|
self.assertEqual('attribute.name', self.ext1.attr)
|
|
self.assertEqual('attribute', self.ext2.attr)
|
|
|
|
def test_entry_point_target(self):
|
|
self.assertEqual('module.name:attribute.name [extra]',
|
|
self.ext1.entry_point_target)
|
|
self.assertEqual('module:attribute',
|
|
self.ext2.entry_point_target)
|