Migrate tox to use testr

This patch moves all unit tests to use
testr as the test runner. This means all
tests need to be subclassed from unittest
to be discovered by testr

Change-Id: I9b0e782be87ea2b5db72b56ddad1d239a83e9ee3
Closes-Bug: #1330204
This commit is contained in:
Abhishek Chanda 2014-08-25 18:18:38 -04:00
parent 4a410fe91a
commit 4f2b64720f
14 changed files with 600 additions and 672 deletions

4
.testr.conf Normal file
View File

@ -0,0 +1,4 @@
[DEFAULT]
test_command=OS_STDOUT_CAPTURE=${OS_STDOUT_CAPTURE:-1} OS_STDERR_CAPTURE=${OS_STDERR_CAPTURE:-1} ${PYTHON:-python} -m subunit.run discover -t ./ ./ $LISTOPT $IDOPTION
test_id_option=--load-list $IDFILE
test_list_option=--list

View File

@ -1,21 +1,25 @@
"""Tests for failure loading callback
"""
from testtools.matchers import GreaterThan
from stevedore import extension
from stevedore.tests import utils
def test_extension_failure_custom_callback():
errors = []
class TestCallback(utils.TestCase):
def test_extension_failure_custom_callback(self):
errors = []
def failure_callback(manager, entrypoint, error):
errors.append((manager, entrypoint, error))
def failure_callback(manager, entrypoint, error):
errors.append((manager, entrypoint, error))
em = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
on_load_failure_callback=failure_callback)
extensions = list(em.extensions)
assert len(extensions) > 0
assert len(errors) == 2
for manager, entrypoint, error in errors:
assert manager is em
assert isinstance(error, (IOError, ImportError))
em = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
on_load_failure_callback=
failure_callback)
extensions = list(em.extensions)
self.assertThat(len(extensions), GreaterThan(0))
self.assertEqual(len(errors), 2)
for manager, entrypoint, error in errors:
self.assertIs(manager, em)
self.assertIsInstance(error, (IOError, ImportError))

View File

@ -1,3 +1,4 @@
from stevedore.tests import utils
from stevedore import dispatch
@ -5,99 +6,86 @@ def check_dispatch(ep, *args, **kwds):
return ep.name == 't2'
def test_dispatch():
class TestDispatch(utils.TestCase):
def check_dispatch(ep, *args, **kwds):
return ep.name == 't2'
def invoke(ep, *args, **kwds):
return (ep.name, args, kwds)
def test_dispatch(self):
em = dispatch.DispatchExtensionManager(
'stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
assert len(em.extensions) == 2
assert set(em.names()) == set(['t1', 't2'])
def invoke(ep, *args, **kwds):
return (ep.name, args, kwds)
results = em.map(check_dispatch,
invoke,
'first',
named='named value',
)
expected = [('t2', ('first',), {'named': 'named value'})]
assert results == expected
em = dispatch.DispatchExtensionManager('stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
self.assertEqual(len(em.extensions), 2)
self.assertEqual(set(em.names()), set(['t1', 't2']))
results = em.map(check_dispatch,
invoke,
'first',
named='named value',
)
expected = [('t2', ('first',), {'named': 'named value'})]
self.assertEqual(results, expected)
def test_dispatch_map_method():
em = dispatch.DispatchExtensionManager(
'stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
def test_dispatch_map_method(self):
em = dispatch.DispatchExtensionManager('stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
results = em.map_method(check_dispatch, 'get_args_and_data',
'first')
assert results == [(('a',), {'b': 'B'}, 'first')]
results = em.map_method(check_dispatch, 'get_args_and_data', 'first')
self.assertEqual(results, [(('a',), {'b': 'B'}, 'first')])
def test_name_dispatch(self):
def test_name_dispatch():
def invoke(ep, *args, **kwds):
return (ep.name, args, kwds)
def invoke(ep, *args, **kwds):
return (ep.name, args, kwds)
em = dispatch.NameDispatchExtensionManager('stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
self.assertEqual(len(em.extensions), 2)
self.assertEqual(set(em.names()), set(['t1', 't2']))
em = dispatch.NameDispatchExtensionManager(
'stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
assert len(em.extensions) == 2
assert set(em.names()) == set(['t1', 't2'])
results = em.map(['t2'], invoke, 'first', named='named value',)
expected = [('t2', ('first',), {'named': 'named value'})]
self.assertEqual(results, expected)
results = em.map(['t2'],
invoke,
'first',
named='named value',
)
expected = [('t2', ('first',), {'named': 'named value'})]
assert results == expected
def test_name_dispatch_ignore_missing(self):
def invoke(ep, *args, **kwds):
return (ep.name, args, kwds)
def test_name_dispatch_ignore_missing():
em = dispatch.NameDispatchExtensionManager(
'stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
def invoke(ep, *args, **kwds):
return (ep.name, args, kwds)
results = em.map(['t3', 't1'], invoke, 'first', named='named value',)
expected = [('t1', ('first',), {'named': 'named value'})]
self.assertEqual(results, expected)
em = dispatch.NameDispatchExtensionManager(
'stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
def test_name_dispatch_map_method(self):
em = dispatch.NameDispatchExtensionManager(
'stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
results = em.map(['t3', 't1'],
invoke,
'first',
named='named value',
)
expected = [('t1', ('first',), {'named': 'named value'})]
assert results == expected
def test_name_dispatch_map_method():
em = dispatch.NameDispatchExtensionManager(
'stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
results = em.map_method(['t3', 't1'], 'get_args_and_data',
'first')
assert results == [(('a',), {'b': 'B'}, 'first')]
results = em.map_method(['t3', 't1'], 'get_args_and_data', 'first')
self.assertEqual(results, [(('a',), {'b': 'B'}, 'first')])

View File

@ -6,65 +6,63 @@ import pkg_resources
from stevedore import driver
from stevedore.tests import test_extension
from stevedore.tests import utils
def test_detect_plugins():
em = driver.DriverManager('stevedore.test.extension', 't1')
names = sorted(em.names())
assert names == ['t1']
class TestCallback(utils.TestCase):
def test_detect_plugins(self):
em = driver.DriverManager('stevedore.test.extension', 't1')
names = sorted(em.names())
self.assertEqual(names, ['t1'])
def test_call(self):
def invoke(ext, *args, **kwds):
return (ext.name, args, kwds)
em = driver.DriverManager('stevedore.test.extension', 't1')
result = em(invoke, 'a', b='C')
self.assertEqual(result, ('t1', ('a',), {'b': 'C'}))
def test_call():
def invoke(ext, *args, **kwds):
return (ext.name, args, kwds)
em = driver.DriverManager('stevedore.test.extension', 't1')
result = em(invoke, 'a', b='C')
assert result == ('t1', ('a',), {'b': 'C'})
def test_driver_property_not_invoked_on_load(self):
em = driver.DriverManager('stevedore.test.extension', 't1',
invoke_on_load=False)
d = em.driver
self.assertIs(d, test_extension.FauxExtension)
def test_driver_property_invoked_on_load(self):
em = driver.DriverManager('stevedore.test.extension', 't1',
invoke_on_load=True)
d = em.driver
self.assertIsInstance(d, test_extension.FauxExtension)
def test_driver_property_not_invoked_on_load():
em = driver.DriverManager('stevedore.test.extension', 't1',
invoke_on_load=False)
d = em.driver
assert d is test_extension.FauxExtension
def test_driver_property_invoked_on_load():
em = driver.DriverManager('stevedore.test.extension', 't1',
invoke_on_load=True)
d = em.driver
assert isinstance(d, test_extension.FauxExtension)
def test_no_drivers():
try:
driver.DriverManager('stevedore.test.extension.none', 't1')
except RuntimeError as err:
assert "No 'stevedore.test.extension.none' driver found" in str(err)
def test_bad_driver():
try:
driver.DriverManager('stevedore.test.extension', 'e2')
except ImportError:
pass
else:
assert False, "No error raised"
def test_multiple_drivers():
# The idea for this test was contributed by clayg:
# https://gist.github.com/clayg/6311348
fep_name = 'stevedore.extension.ExtensionManager._find_entry_points'
with mock.patch(fep_name) as fep:
fep.return_value = [
pkg_resources.EntryPoint.parse('backend = pkg1:driver'),
pkg_resources.EntryPoint.parse('backend = pkg2:driver'),
]
for ep in fep.return_value:
ep.load = lambda *args, **kwds: 'pkg backend'
def test_no_drivers(self):
try:
driver.DriverManager('stevedore.test.multiple_drivers', 'backend')
driver.DriverManager('stevedore.test.extension.none', 't1')
except RuntimeError as err:
assert "Multiple" in str(err), str(err)
fep.assert_called_with('stevedore.test.multiple_drivers')
self.assertIn("No 'stevedore.test.extension.none' driver found",
str(err))
def test_bad_driver(self):
try:
driver.DriverManager('stevedore.test.extension', 'e2')
except ImportError:
pass
else:
self.assertEquals(False, "No error raised")
def test_multiple_drivers(self):
# The idea for this test was contributed by clayg:
# https://gist.github.com/clayg/6311348
fep_name = 'stevedore.extension.ExtensionManager._find_entry_points'
with mock.patch(fep_name) as fep:
fep.return_value = [
pkg_resources.EntryPoint.parse('backend = pkg1:driver'),
pkg_resources.EntryPoint.parse('backend = pkg2:driver'),
]
for ep in fep.return_value:
ep.load = lambda *args, **kwds: 'pkg backend'
try:
driver.DriverManager('stevedore.test.multiple_drivers',
'backend')
except RuntimeError as err:
self.assertIn("Multiple", str(err))
fep.assert_called_with('stevedore.test.multiple_drivers')

View File

@ -1,29 +1,30 @@
from stevedore import enabled
from stevedore.tests import utils
def test_enabled():
def check_enabled(ep):
return ep.name == 't2'
em = enabled.EnabledExtensionManager(
'stevedore.test.extension',
check_enabled,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
assert len(em.extensions) == 1
assert em.names() == ['t2']
class TestEnabled(utils.TestCase):
def test_enabled(self):
def check_enabled(ep):
return ep.name == 't2'
em = enabled.EnabledExtensionManager(
'stevedore.test.extension',
check_enabled,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
self.assertEqual(len(em.extensions), 1)
self.assertEqual(em.names(), ['t2'])
def test_enabled_after_load():
def check_enabled(ext):
return ext.obj and ext.name == 't2'
em = enabled.EnabledExtensionManager(
'stevedore.test.extension',
check_enabled,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
assert len(em.extensions) == 1
assert em.names() == ['t2']
def test_enabled_after_load(self):
def check_enabled(ext):
return ext.obj and ext.name == 't2'
em = enabled.EnabledExtensionManager(
'stevedore.test.extension',
check_enabled,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
self.assertEqual(len(em.extensions), 1)
self.assertEqual(em.names(), ['t2'])

View File

@ -2,26 +2,28 @@
"""
from stevedore.example import fields
from stevedore.tests import utils
def test_simple_items():
f = fields.FieldList(100)
text = ''.join(f.format({'a': 'A', 'b': 'B'}))
expected = '\n'.join([
': a : A',
': b : B',
'',
])
assert text == expected
class TestExampleFields(utils.TestCase):
def test_simple_items(self):
f = fields.FieldList(100)
text = ''.join(f.format({'a': 'A', 'b': 'B'}))
expected = '\n'.join([
': a : A',
': b : B',
'',
])
self.assertEqual(text, expected)
def test_long_item():
f = fields.FieldList(25)
text = ''.join(f.format({'name': 'a value longer than the allowed width'}))
expected = '\n'.join([
': name : a value longer',
' than the allowed',
' width',
'',
])
assert text == expected
def test_long_item(self):
f = fields.FieldList(25)
text = ''.join(f.format({'name':
'a value longer than the allowed width'}))
expected = '\n'.join([
': name : a value longer',
' than the allowed',
' width',
'',
])
self.assertEqual(text, expected)

View File

@ -2,14 +2,16 @@
"""
from stevedore.example import simple
from stevedore.tests import utils
def test_simple_items():
f = simple.Simple(100)
text = ''.join(f.format({'a': 'A', 'b': 'B'}))
expected = '\n'.join([
'a = A',
'b = B',
'',
])
assert text == expected
class TestExampleSimple(utils.TestCase):
def test_simple_items(self):
f = simple.Simple(100)
text = ''.join(f.format({'a': 'A', 'b': 'B'}))
expected = '\n'.join([
'a = A',
'b = B',
'',
])
self.assertEqual(text, expected)

View File

@ -4,6 +4,8 @@
import mock
from stevedore import extension
from stevedore.tests import utils
ALL_NAMES = ['e1', 't1', 't2']
WORKING_NAMES = ['t1', 't2']
@ -23,155 +25,143 @@ class BrokenExtension(object):
raise IOError("Did not create")
def test_detect_plugins():
em = extension.ExtensionManager('stevedore.test.extension')
names = sorted(em.names())
assert names == ALL_NAMES
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_get_by_name():
em = extension.ExtensionManager('stevedore.test.extension')
e = em['t1']
assert e.name == 't1'
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_get_by_name_missing():
em = extension.ExtensionManager('stevedore.test.extension')
try:
em['t3']
except KeyError:
pass
else:
assert False, 'Failed to raise KeyError'
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 pkg_resources
# to find the plugins.
cache = extension.ExtensionManager.ENTRY_POINT_CACHE
cache['stevedore.test.faux'] = []
with mock.patch('pkg_resources.iter_entry_points',
side_effect=
AssertionError('called iter_entry_points')):
em = extension.ExtensionManager('stevedore.test.faux')
names = em.names()
self.assertEqual(names, [])
def test_load_multiple_times_entry_points():
# 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]
assert eps1[0] is eps2[0]
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_load_multiple_times_plugins():
# 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]
assert plugins1[0] is plugins2[0]
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_use_cache():
# If we insert something into the cache of entry points,
# the manager should not have to call into pkg_resources
# to find the plugins.
cache = extension.ExtensionManager.ENTRY_POINT_CACHE
cache['stevedore.test.faux'] = []
with mock.patch('pkg_resources.iter_entry_points',
side_effect=AssertionError('called iter_entry_points')):
em = extension.ExtensionManager('stevedore.test.faux')
names = em.names()
assert names == []
def test_map_arguments(self):
objs = []
def mapped(ext, *args, **kwds):
objs.append((ext, args, kwds))
def test_iterable():
em = extension.ExtensionManager('stevedore.test.extension')
names = sorted(e.name for e in em)
assert names == ALL_NAMES
def test_invoke_on_load():
em = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
assert len(em.extensions) == 2
for e in em.extensions:
assert e.obj.args == ('a',)
assert e.obj.kwds == {'b': 'B'}
def test_map_return_values():
def mapped(ext, *args, **kwds):
return ext.name
em = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
)
results = em.map(mapped)
assert sorted(results) == WORKING_NAMES
def test_map_arguments():
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')
assert len(objs) == 2
names = sorted([o[0].name for o in objs])
assert names == WORKING_NAMES
for o in objs:
assert o[1] == (1, 2)
assert o[2] == {'a': 'A', 'b': 'B'}
def test_map_eats_errors():
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')
assert results == []
def test_map_propagate_exceptions():
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 = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
)
em.map(mapped, 1, 2, a='A', b='B')
assert False
except RuntimeError:
pass
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')
def test_map_errors_when_no_plugins():
em = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
)
results = em.map(mapped, 1, 2, a='A', b='B')
self.assertEqual(results, [])
def mapped(ext, *args, **kwds):
pass
def test_map_propagate_exceptions(self):
def mapped(ext, *args, **kwds):
raise RuntimeError('hard coded error')
em = extension.ExtensionManager('stevedore.test.extension.none',
invoke_on_load=True,
)
try:
em.map(mapped, 1, 2, a='A', b='B')
except RuntimeError as err:
assert 'No stevedore.test.extension.none extensions found' == str(err)
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_method():
em = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
)
def test_map_errors_when_no_plugins(self):
expected_str = 'No stevedore.test.extension.none extensions found'
result = em.map_method('get_args_and_data', 42)
assert set(r[2] for r in result) == set([42])
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 RuntimeError 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]))

View File

@ -1,43 +1,43 @@
from stevedore import hook
from stevedore.tests import utils
def test_hook():
em = hook.HookManager(
'stevedore.test.extension',
't1',
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
assert len(em.extensions) == 1
assert em.names() == ['t1']
class TestHook(utils.TestCase):
def test_hook(self):
em = hook.HookManager(
'stevedore.test.extension',
't1',
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
self.assertEqual(len(em.extensions), 1)
self.assertEqual(em.names(), ['t1'])
def test_get_by_name(self):
em = hook.HookManager(
'stevedore.test.extension',
't1',
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
e_list = em['t1']
self.assertEqual(len(e_list), 1)
e = e_list[0]
self.assertEqual(e.name, 't1')
def test_get_by_name():
em = hook.HookManager(
'stevedore.test.extension',
't1',
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
e_list = em['t1']
assert len(e_list) == 1
e = e_list[0]
assert e.name == 't1'
def test_get_by_name_missing():
em = hook.HookManager(
'stevedore.test.extension',
't1',
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
try:
em['t2']
except KeyError:
pass
else:
assert False, 'Failed to raise KeyError'
def test_get_by_name_missing(self):
em = hook.HookManager(
'stevedore.test.extension',
't1',
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
try:
em['t2']
except KeyError:
pass
else:
assert False, 'Failed to raise KeyError'

View File

@ -1,58 +1,58 @@
from stevedore import named
from stevedore.tests import utils
import mock
def test_named():
em = named.NamedExtensionManager(
'stevedore.test.extension',
names=['t1'],
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
actual = em.names()
assert actual == ['t1']
def test_enabled_before_load():
# Set up the constructor for the FauxExtension to cause an
# AssertionError so the test fails if the class is instantiated,
# which should only happen if it is loaded before the name of the
# extension is compared against the names that should be loaded by
# the manager.
init_name = 'stevedore.tests.test_extension.FauxExtension.__init__'
with mock.patch(init_name) as m:
m.side_effect = AssertionError
class TestNamed(utils.TestCase):
def test_named(self):
em = named.NamedExtensionManager(
'stevedore.test.extension',
# Look for an extension that does not exist so the
# __init__ we mocked should never be invoked.
names=['no-such-extension'],
names=['t1'],
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
actual = em.names()
assert actual == []
self.assertEqual(actual, ['t1'])
def test_enabled_before_load(self):
# Set up the constructor for the FauxExtension to cause an
# AssertionError so the test fails if the class is instantiated,
# which should only happen if it is loaded before the name of the
# extension is compared against the names that should be loaded by
# the manager.
init_name = 'stevedore.tests.test_extension.FauxExtension.__init__'
with mock.patch(init_name) as m:
m.side_effect = AssertionError
em = named.NamedExtensionManager(
'stevedore.test.extension',
# Look for an extension that does not exist so the
# __init__ we mocked should never be invoked.
names=['no-such-extension'],
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
actual = em.names()
self.assertEqual(actual, [])
def test_extensions_listed_in_name_order():
# Since we don't know the "natural" order of the extensions, run
# the test both ways: if the sorting is broken, one of them will
# fail
em = named.NamedExtensionManager(
'stevedore.test.extension',
names=['t1', 't2'],
name_order=True
)
actual = em.names()
assert actual == ['t1', 't2']
def test_extensions_listed_in_name_order(self):
# Since we don't know the "natural" order of the extensions, run
# the test both ways: if the sorting is broken, one of them will
# fail
em = named.NamedExtensionManager(
'stevedore.test.extension',
names=['t1', 't2'],
name_order=True
)
actual = em.names()
self.assertEqual(actual, ['t1', 't2'])
em = named.NamedExtensionManager(
'stevedore.test.extension',
names=['t2', 't1'],
name_order=True
)
actual = em.names()
assert actual == ['t2', 't1']
em = named.NamedExtensionManager(
'stevedore.test.extension',
names=['t2', 't1'],
name_order=True
)
actual = em.names()
self.assertEqual(actual, ['t2', 't1'])

View File

@ -1,10 +1,10 @@
from mock import Mock, sentinel
from nose.tools import raises
from stevedore import (ExtensionManager, NamedExtensionManager, HookManager,
DriverManager, EnabledExtensionManager)
from stevedore.dispatch import (DispatchExtensionManager,
NameDispatchExtensionManager)
from stevedore.extension import Extension
from stevedore.tests import utils
test_extension = Extension('test_extension', None, None, None)
@ -16,263 +16,189 @@ a_driver = Extension('test_driver', mock_entry_point, sentinel.driver_plugin,
# base ExtensionManager
def test_instance_should_use_supplied_extensions():
extensions = [test_extension, test_extension2]
em = ExtensionManager.make_test_instance(extensions)
assert extensions == em.extensions
def test_instance_should_have_default_namespace():
em = ExtensionManager.make_test_instance([])
assert em.namespace
def test_instance_should_use_supplied_namespace():
namespace = 'testing.1.2.3'
em = ExtensionManager.make_test_instance([], namespace=namespace)
assert namespace == em.namespace
def test_extension_name_should_be_listed():
em = ExtensionManager.make_test_instance([test_extension])
assert test_extension.name in em.names()
def test_iterator_should_yield_extension():
em = ExtensionManager.make_test_instance([test_extension])
assert test_extension == next(iter(em))
def test_manager_should_allow_name_access():
em = ExtensionManager.make_test_instance([test_extension])
assert test_extension == em[test_extension.name]
def test_manager_should_call():
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():
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():
def mapped(ext, *args, **kwds):
return ext.name
em = ExtensionManager.make_test_instance([test_extension2,
test_extension])
results = em.map(mapped)
assert sorted(results) == ['another_one', 'test_extension']
def test_manager_should_eat_exceptions():
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')
assert results == []
@raises(RuntimeError)
def test_manager_should_propagate_exceptions():
em = ExtensionManager.make_test_instance([test_extension],
propagate_map_exceptions=True)
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():
extensions = [test_extension, test_extension2]
em = NamedExtensionManager.make_test_instance(extensions)
assert extensions == em.extensions
def test_named_manager_should_have_default_namespace():
em = NamedExtensionManager.make_test_instance([])
assert em.namespace
def test_named_manager_should_use_supplied_namespace():
namespace = 'testing.1.2.3'
em = NamedExtensionManager.make_test_instance([], namespace=namespace)
assert namespace == em.namespace
def test_named_manager_should_populate_names():
extensions = [test_extension, test_extension2]
em = NamedExtensionManager.make_test_instance(extensions)
assert ['test_extension', 'another_one'] == em.names()
# HookManager
def test_hook_manager_should_use_supplied_extensions():
extensions = [test_extension, test_extension2]
em = HookManager.make_test_instance(extensions)
assert extensions == em.extensions
def test_hook_manager_should_be_first_extension_name():
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():
em = HookManager.make_test_instance([test_extension])
assert em.namespace
def test_hook_manager_should_use_supplied_namespace():
namespace = 'testing.1.2.3'
em = HookManager.make_test_instance([test_extension], namespace=namespace)
assert namespace == em.namespace
def test_hook_manager_should_return_named_extensions():
hook1 = Extension('captain', None, None, None)
hook2 = Extension('captain', None, None, None)
em = HookManager.make_test_instance([hook1, hook2])
assert [hook1, hook2] == em['captain']
# DriverManager
def test_driver_manager_should_use_supplied_extension():
em = DriverManager.make_test_instance(a_driver)
assert [a_driver] == em.extensions
def test_driver_manager_should_have_default_namespace():
em = DriverManager.make_test_instance(a_driver)
assert em.namespace
def test_driver_manager_should_use_supplied_namespace():
namespace = 'testing.1.2.3'
em = DriverManager.make_test_instance(a_driver, namespace=namespace)
assert namespace == em.namespace
def test_instance_should_use_driver_name():
em = DriverManager.make_test_instance(a_driver)
assert ['test_driver'] == em.names()
def test_instance_call():
def invoke(ext, *args, **kwds):
return ext.name, args, kwds
em = DriverManager.make_test_instance(a_driver)
result = em(invoke, 'a', b='C')
assert result == ('test_driver', ('a',), {'b': 'C'})
def test_instance_driver_property():
em = DriverManager.make_test_instance(a_driver)
assert sentinel.driver_obj == em.driver
# EnabledExtensionManager
def test_enabled_instance_should_use_supplied_extensions():
extensions = [test_extension, test_extension2]
em = EnabledExtensionManager.make_test_instance(extensions)
assert extensions == em.extensions
# DispatchExtensionManager
def test_dispatch_instance_should_use_supplied_extensions():
extensions = [test_extension, test_extension2]
em = DispatchExtensionManager.make_test_instance(extensions)
assert extensions == em.extensions
def test_dispatch_map_should_invoke_filter_for_extensions():
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():
extensions = [test_extension, test_extension2]
em = NameDispatchExtensionManager.make_test_instance(extensions)
assert extensions == em.extensions
def test_name_dispatch_instance_should_build_extension_name_map():
extensions = [test_extension, test_extension2]
em = NameDispatchExtensionManager.make_test_instance(extensions)
assert test_extension == em.by_name[test_extension.name]
assert test_extension2 == em.by_name[test_extension2.name]
def test_named_dispatch_map_should_invoke_filter_for_extensions():
em = NameDispatchExtensionManager.make_test_instance([test_extension,
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)
func = Mock()
# NameDispatchExtensionManager
def test_name_dispatch_instance_should_use_supplied_extensions(self):
extensions = [test_extension, test_extension2]
em = NameDispatchExtensionManager.make_test_instance(extensions)
args = ('A',)
kw = {'BIGGER': 'Cheese'}
self.assertEqual(extensions, em.extensions)
em.map(['test_extension'], func, *args, **kw)
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])
func.assert_called_once_with(test_extension, *args, **kw)
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)