openstack/stevedore
Code Issues Proposed changes

Apply ruff, ruff-format

Browse Source 
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
Change-Id: Iabae1e4218945e97d4e77df30be1cc9eae9fbc12
This commit is contained in:
Stephen Finucane
2024-08-15 11:00:42 +01:00
parent 2a0a672225
commit bad0502193
32 changed files with 469 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.pre-commit-config.yaml
View File

@@ -17,6 +17,7 @@ repos:
hooks:
- id: hacking
additional_dependencies: []
exclude: '^(doc|releasenotes)/.*$'
- repo: https://github.com/PyCQA/bandit
rev: 1.8.6
hooks:

8
doc/source/conf.py
View File

@@ -84,15 +84,15 @@ latex_documents = [
'stevedore.tex',
'stevedore Documentation',
'DreamHost',
'manual'
),
'manual',
)
]
# -- Options for extlinks extension ---------------------------------------
extlinks = {
'issue': ('https://github.com/dreamhost/stevedore/issues/%s', 'issue '),
'issue': ('https://github.com/dreamhost/stevedore/issues/%s', 'issue ')
}
@@ -101,5 +101,5 @@ extlinks = {
autodoc_default_options = {
'members': None,
'special-members': None,
'show-inheritance': None
'show-inheritance': None,
}

15
releasenotes/source/conf.py
View File

@@ -16,10 +16,7 @@
# Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom ones.
extensions = [
'openstackdocstheme',
'reno.sphinxext',
]
extensions = ['openstackdocstheme', 'reno.sphinxext']
# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']
@@ -67,9 +64,13 @@ html_static_path = ['_static']
# (source start file, target name, title,
# author, documentclass [howto, manual, or own class]).
latex_documents = [
('index', 'stevedoreReleaseNotes.tex',
'stevedore Release Notes Documentation',
'stevedore Developers', 'manual'),
(
'index',
'stevedoreReleaseNotes.tex',
'stevedore Release Notes Documentation',
'stevedore Developers',
'manual',
)
]

4
setup.py
View File

@@ -15,6 +15,4 @@
import setuptools
setuptools.setup(
setup_requires=['pbr>=2.0.0'],
pbr=True)
setuptools.setup(setup_requires=['pbr>=2.0.0'], pbr=True)

30
stevedore/_cache.py
View File

@@ -36,8 +36,9 @@ def _get_cache_dir():
# Linux, Unix, AIX, etc.
if os.name == 'posix' and sys.platform != 'darwin':
# use ~/.cache if empty OR not set
base_path = os.environ.get("XDG_CACHE_HOME", None) \
or os.path.expanduser('~/.cache')
base_path = os.environ.get(
"XDG_CACHE_HOME", None
) or os.path.expanduser('~/.cache')
return os.path.join(base_path, 'python-entrypoints')
# Mac OS
@@ -46,8 +47,9 @@ def _get_cache_dir():
# Windows (hopefully)
else:
base_path = os.environ.get('LOCALAPPDATA', None) \
or os.path.expanduser('~\\AppData\\Local')
base_path = os.environ.get('LOCALAPPDATA', None) or os.path.expanduser(
'~\\AppData\\Local'
)
return os.path.join(base_path, 'Python Entry Points')
@@ -82,12 +84,8 @@ def _hash_settings_for_path(path):
paths.append((entry, mtime))
for ep_file in itertools.chain(
glob.iglob(os.path.join(entry,
'*.dist-info',
'entry_points.txt')),
glob.iglob(os.path.join(entry,
'*.egg-info',
'entry_points.txt'))
glob.iglob(os.path.join(entry, '*.dist-info', 'entry_points.txt')),
glob.iglob(os.path.join(entry, '*.egg-info', 'entry_points.txt')),
):
mtime = _get_mtime(ep_file)
h.update(ep_file.encode('utf-8'))
@@ -132,7 +130,6 @@ def _build_cacheable_data():
class Cache:
def __init__(self, cache_dir=None):
if cache_dir is None:
cache_dir = _get_cache_dir()
@@ -143,8 +140,12 @@ class Cache:
# Caching can be disabled by either placing .disable file into the
# target directory or when python executable is under /tmp (this is the
# case when executed from ansible)
if any([os.path.isfile(os.path.join(self._dir, '.disable')),
sys.executable[0:4] == '/tmp']): # nosec B108
if any(
[
os.path.isfile(os.path.join(self._dir, '.disable')),
sys.executable[0:4] == '/tmp', # nosec B108,
]
):
self._disable_caching = True
def _get_data_for_path(self, path):
@@ -196,8 +197,7 @@ class Cache:
for name, ep in self.get_group_named(group, path=path).items():
if name == name:
return ep
raise ValueError('No entrypoint {!r} in group {!r}'.format(
group, name))
raise ValueError(f'No entrypoint {group!r} in group {name!r}')
_c = Cache()

32
stevedore/dispatch.py
View File

@@ -79,7 +79,7 @@ class DispatchExtensionManager(EnabledExtensionManager):
"""
if not self.extensions:
# FIXME: Use a more specific exception class here.
raise NoMatches('No %s extensions found' % self.namespace)
raise NoMatches(f'No {self.namespace} extensions found')
response = []
for e in self.extensions:
if filter_func(e, *args, **kwds):
@@ -107,8 +107,13 @@ class DispatchExtensionManager(EnabledExtensionManager):
:param kwds: Keyword arguments to pass to method
:returns: List of values returned from methods
"""
return self.map(filter_func, self._call_extension_method,
method_name, *args, **kwds)
return self.map(
filter_func,
self._call_extension_method,
method_name,
*args,
**kwds,
)
class NameDispatchExtensionManager(DispatchExtensionManager):
@@ -152,11 +157,17 @@ class NameDispatchExtensionManager(DispatchExtensionManager):
"""
def __init__(self, namespace, check_func, invoke_on_load=False,
invoke_args=(), invoke_kwds={},
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False):
def __init__(
self,
namespace,
check_func,
invoke_on_load=False,
invoke_args=(),
invoke_kwds={},
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False,
):
super().__init__(
namespace=namespace,
check_func=check_func,
@@ -225,5 +236,6 @@ class NameDispatchExtensionManager(DispatchExtensionManager):
:param kwds: Keyword arguments to pass to method
:returns: List of values returned from methods
"""
return self.map(names, self._call_extension_method,
method_name, *args, **kwds)
return self.map(
names, self._call_extension_method, method_name, *args, **kwds
)

57
stevedore/driver.py
View File

@@ -44,13 +44,20 @@ class DriverManager(NamedExtensionManager):
:type warn_on_missing_entrypoint: bool
"""
def __init__(self, namespace, name,
invoke_on_load=False, invoke_args=(), invoke_kwds={},
on_load_failure_callback=None,
verify_requirements=False,
warn_on_missing_entrypoint=True):
on_load_failure_callback = on_load_failure_callback \
or self._default_on_load_failure
def __init__(
self,
namespace,
name,
invoke_on_load=False,
invoke_args=(),
invoke_kwds={},
on_load_failure_callback=None,
verify_requirements=False,
warn_on_missing_entrypoint=True,
):
on_load_failure_callback = (
on_load_failure_callback or self._default_on_load_failure
)
super().__init__(
namespace=namespace,
names=[name],
@@ -59,7 +66,7 @@ class DriverManager(NamedExtensionManager):
invoke_kwds=invoke_kwds,
on_load_failure_callback=on_load_failure_callback,
verify_requirements=verify_requirements,
warn_on_missing_entrypoint=warn_on_missing_entrypoint
warn_on_missing_entrypoint=warn_on_missing_entrypoint,
)
@staticmethod
@@ -67,10 +74,14 @@ class DriverManager(NamedExtensionManager):
raise
@classmethod
def make_test_instance(cls, extension, namespace='TESTING',
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False):
def make_test_instance(
cls,
extension,
namespace='TESTING',
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False,
):
"""Construct a test DriverManager
Test instances are passed a list of extensions to work from rather
@@ -99,10 +110,12 @@ class DriverManager(NamedExtensionManager):
"""
o = super().make_test_instance(
[extension], namespace=namespace,
[extension],
namespace=namespace,
propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback,
verify_requirements=verify_requirements)
verify_requirements=verify_requirements,
)
return o
def _init_plugins(self, extensions):
@@ -110,14 +123,18 @@ class DriverManager(NamedExtensionManager):
if not self.extensions:
name = self._names[0]
raise NoMatches('No %r driver found, looking for %r' %
(self.namespace, name))
raise NoMatches(
f'No {self.namespace!r} driver found, looking for {name!r}'
)
if len(self.extensions) > 1:
discovered_drivers = ','.join(e.entry_point_target
for e in self.extensions)
discovered_drivers = ','.join(
e.entry_point_target for e in self.extensions
)
raise MultipleMatches('Multiple %r drivers found: %s' %
(self.namespace, discovered_drivers))
raise MultipleMatches(
f'Multiple {self.namespace!r} drivers found: '
f'{discovered_drivers}'
)
def __call__(self, func, *args, **kwds):
"""Invokes func() for the single loaded extension.

24
stevedore/enabled.py
View File

@@ -56,11 +56,17 @@ class EnabledExtensionManager(ExtensionManager):
"""
def __init__(self, namespace, check_func, invoke_on_load=False,
invoke_args=(), invoke_kwds={},
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False,):
def __init__(
self,
namespace,
check_func,
invoke_on_load=False,
invoke_args=(),
invoke_kwds={},
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False,
):
self.check_func = check_func
super().__init__(
namespace,
@@ -72,11 +78,11 @@ class EnabledExtensionManager(ExtensionManager):
verify_requirements=verify_requirements,
)
def _load_one_plugin(self, ep, invoke_on_load, invoke_args, invoke_kwds,
verify_requirements):
def _load_one_plugin(
self, ep, invoke_on_load, invoke_args, invoke_kwds, verify_requirements
):
ext = super()._load_one_plugin(
ep, invoke_on_load, invoke_args, invoke_kwds,
verify_requirements,
ep, invoke_on_load, invoke_args, invoke_kwds, verify_requirements
)
if ext and not self.check_func(ext):
LOG.debug('ignoring extension %r', ep.name)

3
stevedore/example/base.py
View File

@@ -16,8 +16,7 @@ import abc
class FormatterBase(metaclass=abc.ABCMeta):
"""Base class for example plugin used in the tutorial.
"""
"""Base class for example plugin used in the tutorial."""
def __init__(self, max_width=60):
self.max_width = max_width

16
stevedore/example/load_as_driver.py
View File

@@ -20,24 +20,14 @@ from stevedore import driver
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument(
'format',
nargs='?',
default='simple',
help='the output format',
'format', nargs='?', default='simple', help='the output format'
)
parser.add_argument(
'--width',
default=60,
type=int,
help='maximum output width for text',
'--width', default=60, type=int, help='maximum output width for text'
)
parsed_args = parser.parse_args()
data = {
'a': 'A',
'b': 'B',
'long': 'word ' * 80,
}
data = {'a': 'A', 'b': 'B', 'long': 'word ' * 80}
mgr = driver.DriverManager(
namespace='stevedore.example.formatter',

11
stevedore/example/load_as_extension.py
View File

@@ -20,18 +20,11 @@ from stevedore import extension
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument(
'--width',
default=60,
type=int,
help='maximum output width for text',
'--width', default=60, type=int, help='maximum output width for text'
)
parsed_args = parser.parse_args()
data = {
'a': 'A',
'b': 'B',
'long': 'word ' * 80,
}
data = {'a': 'A', 'b': 'B', 'long': 'word ' * 80}
mgr = extension.ExtensionManager(
namespace='stevedore.example.formatter',

36
stevedore/example/setup.py
View File

@@ -18,41 +18,31 @@ from setuptools import setup
setup(
name='stevedore-examples',
version='1.0',
description='Demonstration package for stevedore',
author='Doug Hellmann',
author_email='doug@doughellmann.com',
url='http://opendev.org/openstack/stevedore',
classifiers=['Development Status :: 3 - Alpha',
'License :: OSI Approved :: Apache Software License',
'Programming Language :: Python',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.5',
'Intended Audience :: Developers',
'Environment :: Console',
],
classifiers=[
'Development Status :: 3 - Alpha',
'License :: OSI Approved :: Apache Software License',
'Programming Language :: Python',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.5',
'Intended Audience :: Developers',
'Environment :: Console',
],
platforms=['Any'],
scripts=[],
provides=['stevedore.examples',
],
provides=['stevedore.examples'],
packages=find_packages(),
include_package_data=True,
entry_points={
'stevedore.example.formatter': [
'simple = stevedore.example.simple:Simple',
'plain = stevedore.example.simple:Simple',
],
]
},
zip_safe=False,
)

5
stevedore/example/simple.py
View File

@@ -25,8 +25,5 @@ class Simple(base.FormatterBase):
:type data: dict(str:?)
"""
for name, value in sorted(data.items()):
line = '{name} = {value}\n'.format(
name=name,
value=value,
)
line = f'{name} = {value}\n'
yield line

5
stevedore/example2/fields.py
View File

@@ -37,10 +37,7 @@ class FieldList(base.FormatterBase):
:type data: dict(str:?)
"""
for name, value in sorted(data.items()):
full_text = ': {name} : {value}'.format(
name=name,
value=value,
)
full_text = f': {name} : {value}'
wrapped_text = textwrap.fill(
full_text,
initial_indent='',

38
stevedore/example2/setup.py
View File

@@ -18,40 +18,30 @@ from setuptools import setup
setup(
name='stevedore-examples2',
version='1.0',
description='Demonstration package for stevedore',
author='Doug Hellmann',
author_email='doug@doughellmann.com',
url='http://opendev.org/openstack/stevedore',
classifiers=['Development Status :: 3 - Alpha',
'License :: OSI Approved :: Apache Software License',
'Programming Language :: Python',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.5',
'Intended Audience :: Developers',
'Environment :: Console',
],
classifiers=[
'Development Status :: 3 - Alpha',
'License :: OSI Approved :: Apache Software License',
'Programming Language :: Python',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.5',
'Intended Audience :: Developers',
'Environment :: Console',
],
platforms=['Any'],
scripts=[],
provides=['stevedore.examples2',
],
provides=['stevedore.examples2'],
packages=find_packages(),
include_package_data=True,
entry_points={
'stevedore.example.formatter': [
'field = stevedore.example2.fields:FieldList',
],
'field = stevedore.example2.fields:FieldList'
]
},
zip_safe=False,
)

96
stevedore/extension.py
View File

@@ -10,8 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
"""ExtensionManager
"""
"""ExtensionManager"""
import logging
import operator
@@ -104,28 +103,35 @@ class ExtensionManager:
:type verify_requirements: bool
"""
def __init__(self, namespace,
invoke_on_load=False,
invoke_args=(),
invoke_kwds={},
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False):
def __init__(
self,
namespace,
invoke_on_load=False,
invoke_args=(),
invoke_kwds={},
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False,
):
self._init_attributes(
namespace,
propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback)
extensions = self._load_plugins(invoke_on_load,
invoke_args,
invoke_kwds,
verify_requirements)
on_load_failure_callback=on_load_failure_callback,
)
extensions = self._load_plugins(
invoke_on_load, invoke_args, invoke_kwds, verify_requirements
)
self._init_plugins(extensions)
@classmethod
def make_test_instance(cls, extensions, namespace='TESTING',
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False):
def make_test_instance(
cls,
extensions,
namespace='TESTING',
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False,
):
"""Construct a test ExtensionManager
Test instances are passed a list of extensions to work from rather
@@ -154,14 +160,20 @@ class ExtensionManager:
"""
o = cls.__new__(cls)
o._init_attributes(namespace,
propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback)
o._init_attributes(
namespace,
propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback,
)
o._init_plugins(extensions)
return o
def _init_attributes(self, namespace, propagate_map_exceptions=False,
on_load_failure_callback=None):
def _init_attributes(
self,
namespace,
propagate_map_exceptions=False,
on_load_failure_callback=None,
):
self.namespace = namespace
self.propagate_map_exceptions = propagate_map_exceptions
self._on_load_failure_callback = on_load_failure_callback
@@ -197,18 +209,20 @@ class ExtensionManager:
"""Return the list of entry points names for this namespace."""
return list(map(operator.attrgetter("name"), self.list_entry_points()))
def _load_plugins(self, invoke_on_load, invoke_args, invoke_kwds,
verify_requirements):
def _load_plugins(
self, invoke_on_load, invoke_args, invoke_kwds, verify_requirements
):
extensions = []
for ep in self.list_entry_points():
LOG.debug('found extension %r', ep)
try:
ext = self._load_one_plugin(ep,
invoke_on_load,
invoke_args,
invoke_kwds,
verify_requirements,
)
ext = self._load_one_plugin(
ep,
invoke_on_load,
invoke_args,
invoke_kwds,
verify_requirements,
)
if ext:
extensions.append(ext)
except (KeyboardInterrupt, AssertionError):
@@ -224,12 +238,17 @@ class ExtensionManager:
# enough to debug that. If debug logging is
# enabled for our logger, provide the full
# traceback.
LOG.error('Could not load %r: %s', ep.name, err,
exc_info=LOG.isEnabledFor(logging.DEBUG))
LOG.error(
'Could not load %r: %s',
ep.name,
err,
exc_info=LOG.isEnabledFor(logging.DEBUG),
)
return extensions
def _load_one_plugin(self, ep, invoke_on_load, invoke_args, invoke_kwds,
verify_requirements):
def _load_one_plugin(
self, ep, invoke_on_load, invoke_args, invoke_kwds, verify_requirements
):
# NOTE(dhellmann): Using require=False is deprecated in
# setuptools 11.3.
if hasattr(ep, 'resolve') and hasattr(ep, 'require'):
@@ -273,7 +292,7 @@ class ExtensionManager:
"""
if not self.extensions:
# FIXME: Use a more specific exception class here.
raise NoMatches('No %s extensions found' % self.namespace)
raise NoMatches(f'No {self.namespace} extensions found')
response = []
for e in self.extensions:
self._invoke_one_plugin(response.append, func, e, args, kwds)
@@ -302,8 +321,9 @@ class ExtensionManager:
:param kwds: Keyword arguments to pass to method
:returns: List of values returned from methods
"""
return self.map(self._call_extension_method,
method_name, *args, **kwds)
return self.map(
self._call_extension_method, method_name, *args, **kwds
)
def _invoke_one_plugin(self, response_callback, func, e, args, kwds):
try:

41
stevedore/hook.py
View File

@@ -47,15 +47,21 @@ class HookManager(NamedExtensionManager):
"""
def __init__(self, namespace, name,
invoke_on_load=False, invoke_args=(), invoke_kwds={},
on_load_failure_callback=None,
verify_requirements=False,
on_missing_entrypoints_callback=None,
# NOTE(dhellmann): This default is different from the
# base class because for hooks it is less likely to
# be an error to have no entry points present.
warn_on_missing_entrypoint=False):
def __init__(
self,
namespace,
name,
invoke_on_load=False,
invoke_args=(),
invoke_kwds={},
on_load_failure_callback=None,
verify_requirements=False,
on_missing_entrypoints_callback=None,
# NOTE(dhellmann): This default is different from the
# base class because for hooks it is less likely to
# be an error to have no entry points present.
warn_on_missing_entrypoint=False,
):
super().__init__(
namespace,
[name],
@@ -68,13 +74,20 @@ class HookManager(NamedExtensionManager):
warn_on_missing_entrypoint=warn_on_missing_entrypoint,
)
def _init_attributes(self, namespace, names, name_order=False,
propagate_map_exceptions=False,
on_load_failure_callback=None):
def _init_attributes(
self,
namespace,
names,
name_order=False,
propagate_map_exceptions=False,
on_load_failure_callback=None,
):
super()._init_attributes(
namespace, names,
namespace,
names,
propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback)
on_load_failure_callback=on_load_failure_callback,
)
self._name = names[0]
def __getitem__(self, name):

91
stevedore/named.py
View File

@@ -64,35 +64,49 @@ class NamedExtensionManager(ExtensionManager):
"""
def __init__(self, namespace, names,
invoke_on_load=False, invoke_args=(), invoke_kwds={},
name_order=False, propagate_map_exceptions=False,
on_load_failure_callback=None,
on_missing_entrypoints_callback=None,
verify_requirements=False,
warn_on_missing_entrypoint=True):
def __init__(
self,
namespace,
names,
invoke_on_load=False,
invoke_args=(),
invoke_kwds={},
name_order=False,
propagate_map_exceptions=False,
on_load_failure_callback=None,
on_missing_entrypoints_callback=None,
verify_requirements=False,
warn_on_missing_entrypoint=True,
):
self._init_attributes(
namespace, names, name_order=name_order,
namespace,
names,
name_order=name_order,
propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback)
extensions = self._load_plugins(invoke_on_load,
invoke_args,
invoke_kwds,
verify_requirements)
on_load_failure_callback=on_load_failure_callback,
)
extensions = self._load_plugins(
invoke_on_load, invoke_args, invoke_kwds, verify_requirements
)
self._missing_names = set(names) - {e.name for e in extensions}
if self._missing_names:
if on_missing_entrypoints_callback:
on_missing_entrypoints_callback(self._missing_names)
elif warn_on_missing_entrypoint:
LOG.warning('Could not load %s' %
', '.join(self._missing_names))
LOG.warning(
'Could not load {}'.format(', '.join(self._missing_names))
)
self._init_plugins(extensions)
@classmethod
def make_test_instance(cls, extensions, namespace='TESTING',
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False):
def make_test_instance(
cls,
extensions,
namespace='TESTING',
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False,
):
"""Construct a test NamedExtensionManager
Test instances are passed a list of extensions to use rather than
@@ -122,18 +136,28 @@ class NamedExtensionManager(ExtensionManager):
o = cls.__new__(cls)
names = [e.name for e in extensions]
o._init_attributes(namespace, names,
propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback)
o._init_attributes(
namespace,
names,
propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback,
)
o._init_plugins(extensions)
return o
def _init_attributes(self, namespace, names, name_order=False,
propagate_map_exceptions=False,
on_load_failure_callback=None):
def _init_attributes(
self,
namespace,
names,
name_order=False,
propagate_map_exceptions=False,
on_load_failure_callback=None,
):
super()._init_attributes(
namespace, propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback)
namespace,
propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback,
)
self._names = names
self._missing_names = set()
@@ -143,17 +167,18 @@ class NamedExtensionManager(ExtensionManager):
super()._init_plugins(extensions)
if self._name_order:
self.extensions = [self[n] for n in self._names
if n not in self._missing_names]
self.extensions = [
self[n] for n in self._names if n not in self._missing_names
]
def _load_one_plugin(self, ep, invoke_on_load, invoke_args, invoke_kwds,
verify_requirements):
def _load_one_plugin(
self, ep, invoke_on_load, invoke_args, invoke_kwds, verify_requirements
):
# Check the name before going any further to prevent
# undesirable code from being loaded at all if we are not
# going to use it.
if ep.name not in self._names:
return None
return super()._load_one_plugin(
ep, invoke_on_load, invoke_args, invoke_kwds,
verify_requirements,
ep, invoke_on_load, invoke_args, invoke_kwds, verify_requirements
)

25
stevedore/sphinxext.py
View File

@@ -33,8 +33,7 @@ def _simple_list(mgr):
ext = mgr[name]
doc = _get_docstring(ext.plugin) or '\n'
summary = doc.splitlines()[0].strip()
yield (f'* {ext.name} -- {summary}',
ext.module_name)
yield (f'* {ext.name} -- {summary}', ext.module_name)
def _detailed_list(mgr, over='', under='-', titlecase=False):
@@ -54,9 +53,8 @@ def _detailed_list(mgr, over='', under='-', titlecase=False):
yield (doc, ext.module_name)
else:
yield (
'.. warning:: No documentation found for {} in {}'.format(
ext.name, ext.entry_point_target,
),
f'.. warning:: No documentation found for {ext.name} in '
f'{ext.entry_point_target}',
ext.module_name,
)
yield ('\n', ext.module_name)
@@ -77,7 +75,7 @@ class ListPluginsDirective(rst.Directive):
def run(self):
namespace = ' '.join(self.content).strip()
LOG.info('documenting plugins from %r' % namespace)
LOG.info(f'documenting plugins from {namespace!r}')
overline_style = self.options.get('overline-style', '')
underline_style = self.options.get('underline-style', '=')
@@ -85,8 +83,7 @@ class ListPluginsDirective(rst.Directive):
LOG.warning(f'Failed to load {ep.module}: {err}')
mgr = extension.ExtensionManager(
namespace,
on_load_failure_callback=report_load_failure,
namespace, on_load_failure_callback=report_load_failure
)
result = ViewList()
@@ -95,8 +92,11 @@ class ListPluginsDirective(rst.Directive):
if 'detailed' in self.options:
data = _detailed_list(
mgr, over=overline_style, under=underline_style,
titlecase=titlecase)
mgr,
over=overline_style,
under=underline_style,
titlecase=titlecase,
)
else:
data = _simple_list(mgr)
for text, source in data:
@@ -114,7 +114,4 @@ class ListPluginsDirective(rst.Directive):
def setup(app):
LOG.info('loading stevedore.sphinxext')
app.add_directive('list-plugins', ListPluginsDirective)
return {
'parallel_read_safe': True,
'parallel_write_safe': True,
}
return {'parallel_read_safe': True, 'parallel_write_safe': True}

22
stevedore/tests/manager.py
View File

@@ -48,20 +48,20 @@ class TestExtensionManager(extension.ExtensionManager):
"""
def __init__(self, extensions,
namespace='test',
invoke_on_load=False,
invoke_args=(),
invoke_kwds={}):
super().__init__(namespace,
invoke_on_load,
invoke_args,
invoke_kwds,
)
def __init__(
self,
extensions,
namespace='test',
invoke_on_load=False,
invoke_args=(),
invoke_kwds={},
):
super().__init__(namespace, invoke_on_load, invoke_args, invoke_kwds)
self.extensions = extensions
warnings.warn(
'TestExtesionManager has been replaced by make_test_instance()',
DeprecationWarning)
DeprecationWarning,
)
def _load_plugins(self, *args, **kwds):
return []

7
stevedore/tests/test_cache.py
View File

@@ -10,8 +10,8 @@
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for stevedore._cache
"""
"""Tests for stevedore._cache"""
import sys
from unittest import mock
@@ -21,7 +21,6 @@ from stevedore.tests import utils
class TestCache(utils.TestCase):
def test_disable_caching_executable(self):
"""Test caching is disabled if python interpreter is located under /tmp
directory (Ansible)
@@ -39,7 +38,7 @@ class TestCache(utils.TestCase):
with mock.patch('os.path.isfile') as mock_path:
mock_path.return_value = True
sot = _cache.Cache()
mock_path.assert_called_with('%s/.disable' % cache_dir)
mock_path.assert_called_with(f'{cache_dir}/.disable')
self.assertTrue(sot._disable_caching)
mock_path.return_value = False

27
stevedore/tests/test_callback.py
View File

@@ -10,8 +10,8 @@
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for failure loading callback
"""
"""Tests for failure loading callback"""
from unittest import mock
from testtools.matchers import GreaterThan
@@ -28,10 +28,11 @@ class TestCallback(utils.TestCase):
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)
em = extension.ExtensionManager(
'stevedore.test.extension',
invoke_on_load=True,
on_load_failure_callback=failure_callback,
)
extensions = list(em.extensions)
self.assertTrue(len(extensions), GreaterThan(0))
self.assertEqual(len(errors), 2)
@@ -46,11 +47,11 @@ class TestCallback(utils.TestCase):
def callback(names):
errors.update(names)
load_fn.return_value = [
extension.Extension('foo', None, None, None)
]
named.NamedExtensionManager('stevedore.test.extension',
names=['foo', 'bar'],
invoke_on_load=True,
on_missing_entrypoints_callback=callback)
load_fn.return_value = [extension.Extension('foo', None, None, None)]
named.NamedExtensionManager(
'stevedore.test.extension',
names=['foo', 'bar'],
invoke_on_load=True,
on_missing_entrypoints_callback=callback,
)
self.assertEqual(errors, {'bar'})

52
stevedore/tests/test_dispatch.py
View File

@@ -23,58 +23,54 @@ class TestDispatch(utils.TestCase):
return ep.name == 't2'
def test_dispatch(self):
def invoke(ep, *args, **kwds):
return (ep.name, args, kwds)
em = dispatch.DispatchExtensionManager('stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
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()), {'t1', 't2'})
results = em.map(check_dispatch,
invoke,
'first',
named='named value',
)
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(self):
em = dispatch.DispatchExtensionManager('stevedore.test.extension',
lambda *args, **kwds: True,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
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')
self.assertEqual(results, [(('a',), {'b': 'B'}, 'first')])
def test_name_dispatch(self):
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'},
)
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()), {'t1', 't2'})
results = em.map(['t2'], invoke, 'first', named='named value',)
results = em.map(['t2'], invoke, 'first', named='named value')
expected = [('t2', ('first',), {'named': 'named value'})]
self.assertEqual(results, expected)
def test_name_dispatch_ignore_missing(self):
def invoke(ep, *args, **kwds):
return (ep.name, args, kwds)
@@ -86,7 +82,7 @@ class TestDispatch(utils.TestCase):
invoke_kwds={'b': 'B'},
)
results = em.map(['t3', 't1'], invoke, 'first', named='named value',)
results = em.map(['t3', 't1'], invoke, 'first', named='named value')
expected = [('t1', ('first',), {'named': 'named value'})]
self.assertEqual(results, expected)

25
stevedore/tests/test_driver.py
View File

@@ -10,8 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for stevedore.extension
"""
"""Tests for stevedore.extension"""
import importlib.metadata as importlib_metadata
@@ -31,19 +30,22 @@ class TestCallback(utils.TestCase):
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_driver_property_not_invoked_on_load(self):
em = driver.DriverManager('stevedore.test.extension', 't1',
invoke_on_load=False)
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)
em = driver.DriverManager(
'stevedore.test.extension', 't1', invoke_on_load=True
)
d = em.driver
self.assertIsInstance(d, test_extension.FauxExtension)
@@ -51,8 +53,9 @@ class TestCallback(utils.TestCase):
try:
driver.DriverManager('stevedore.test.extension.none', 't1')
except exception.NoMatches as err:
self.assertIn("No 'stevedore.test.extension.none' driver found",
str(err))
self.assertIn(
"No 'stevedore.test.extension.none' driver found", str(err)
)
def test_bad_driver(self):
try:
@@ -69,14 +72,16 @@ class TestCallback(utils.TestCase):
extension.Extension(
'backend',
importlib_metadata.EntryPoint(
'backend', 'pkg1:driver', 'backend'),
'backend', 'pkg1:driver', 'backend'
),
'pkg backend',
None,
),
extension.Extension(
'backend',
importlib_metadata.EntryPoint(
'backend', 'pkg2:driver', 'backend'),
'backend', 'pkg2:driver', 'backend'
),
'pkg backend',
None,
),

2
stevedore/tests/test_enabled.py
View File

@@ -18,6 +18,7 @@ class TestEnabled(utils.TestCase):
def test_enabled(self):
def check_enabled(ep):
return ep.name == 't2'
em = enabled.EnabledExtensionManager(
'stevedore.test.extension',
check_enabled,
@@ -31,6 +32,7 @@ class TestEnabled(utils.TestCase):
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,

28
stevedore/tests/test_example_fields.py
View File

@@ -10,8 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for stevedore.example2.fields
"""
"""Tests for stevedore.example2.fields"""
from stevedore.example2 import fields
from stevedore.tests import utils
@@ -21,21 +20,20 @@ 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',
'',
])
expected = '\n'.join([': a : A', ': b : B', ''])
self.assertEqual(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',
'',
])
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)

9
stevedore/tests/test_example_simple.py
View File

@@ -10,8 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for stevedore.example.simple
"""
"""Tests for stevedore.example.simple"""
from stevedore.example import simple
from stevedore.tests import utils
@@ -21,9 +20,5 @@ 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',
'',
])
expected = '\n'.join(['a = A', 'b = B', ''])
self.assertEqual(text, expected)

98
stevedore/tests/test_extension.py
View File

@@ -10,8 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for stevedore.extension
"""
"""Tests for stevedore.extension"""
import importlib.metadata as importlib_metadata
import operator
@@ -54,8 +53,9 @@ class TestCallback(utils.TestCase):
def test_list_entry_points(self):
em = extension.ExtensionManager('stevedore.test.extension')
n = em.list_entry_points()
self.assertEqual({'e1', 'e2', 't1', 't2'},
set(map(operator.attrgetter("name"), n)))
self.assertEqual(
{'e1', 'e2', 't1', 't2'}, set(map(operator.attrgetter("name"), n))
)
self.assertEqual(4, len(n))
def test_list_entry_points_names(self):
@@ -101,9 +101,10 @@ class TestCallback(utils.TestCase):
# 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')):
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, [])
@@ -114,11 +115,12 @@ class TestCallback(utils.TestCase):
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'},
)
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',))
@@ -128,9 +130,9 @@ class TestCallback(utils.TestCase):
def mapped(ext, *args, **kwds):
return ext.name
em = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
)
em = extension.ExtensionManager(
'stevedore.test.extension', invoke_on_load=True
)
results = em.map(mapped)
self.assertEqual(sorted(results), WORKING_NAMES)
@@ -140,9 +142,9 @@ class TestCallback(utils.TestCase):
def mapped(ext, *args, **kwds):
objs.append((ext, args, kwds))
em = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
)
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])
@@ -155,9 +157,9 @@ class TestCallback(utils.TestCase):
def mapped(ext, *args, **kwds):
raise RuntimeError('hard coded error')
em = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
)
em = extension.ExtensionManager(
'stevedore.test.extension', invoke_on_load=True
)
results = em.map(mapped, 1, 2, a='A', b='B')
self.assertEqual(results, [])
@@ -165,10 +167,11 @@ class TestCallback(utils.TestCase):
def mapped(ext, *args, **kwds):
raise RuntimeError('hard coded error')
em = extension.ExtensionManager('stevedore.test.extension',
invoke_on_load=True,
propagate_map_exceptions=True
)
em = extension.ExtensionManager(
'stevedore.test.extension',
invoke_on_load=True,
propagate_map_exceptions=True,
)
try:
em.map(mapped, 1, 2, a='A', b='B')
@@ -182,18 +185,18 @@ class TestCallback(utils.TestCase):
def mapped(ext, *args, **kwds):
pass
em = extension.ExtensionManager('stevedore.test.extension.none',
invoke_on_load=True,
)
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,
)
em = extension.ExtensionManager(
'stevedore.test.extension', invoke_on_load=True
)
result = em.map_method('get_args_and_data', 42)
self.assertEqual({r[2] for r in result}, {42})
@@ -213,14 +216,16 @@ class TestLoadRequirementsNewSetuptools(utils.TestCase):
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.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.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()
@@ -234,23 +239,24 @@ class TestLoadRequirementsOldSetuptools(utils.TestCase):
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.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.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',
'name', 'module.name:attribute.name [extra]', 'group_name'
),
mock.Mock(),
None,
@@ -258,7 +264,7 @@ class TestExtensionProperties(utils.TestCase):
self.ext2 = extension.Extension(
'name',
importlib_metadata.EntryPoint(
'name', 'module:attribute', 'group_name',
'name', 'module:attribute', 'group_name'
),
mock.Mock(),
None,
@@ -273,7 +279,7 @@ class TestExtensionProperties(utils.TestCase):
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)
self.assertEqual(
'module.name:attribute.name [extra]', self.ext1.entry_point_target
)
self.assertEqual('module:attribute', self.ext2.entry_point_target)

8
stevedore/tests/test_named.py
View File

@@ -54,17 +54,13 @@ class TestNamed(utils.TestCase):
# 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
'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
'stevedore.test.extension', names=['t2', 't1'], name_order=True
)
actual = em.names()
self.assertEqual(actual, ['t2', 't1'])

25
stevedore/tests/test_sphinxext.py
View File

@@ -9,8 +9,7 @@
# 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 the sphinx extension
"""
"""Tests for the sphinx extension"""
import importlib.metadata as importlib_metadata
@@ -24,14 +23,11 @@ def _make_ext(name, docstring):
pass
inner.__doc__ = docstring
m1 = importlib_metadata.EntryPoint(
name, f'{name}_module:{name}', 'group',
)
m1 = importlib_metadata.EntryPoint(name, f'{name}_module:{name}', 'group')
return extension.Extension(name, m1, inner, None)
class TestSphinxExt(utils.TestCase):
def setUp(self):
super().setUp()
self.exts = [
@@ -54,12 +50,7 @@ class TestSphinxExt(utils.TestCase):
ext = [_make_ext('nodoc', None)]
em = extension.ExtensionManager.make_test_instance(ext)
results = list(sphinxext._simple_list(em))
self.assertEqual(
[
('* nodoc -- ', 'nodoc_module'),
],
results,
)
self.assertEqual([('* nodoc -- ', 'nodoc_module')], results)
def test_detailed_list(self):
results = list(sphinxext._detailed_list(self.em))
@@ -108,9 +99,13 @@ class TestSphinxExt(utils.TestCase):
('nodoc', 'nodoc_module'),
('-----', 'nodoc_module'),
('\n', 'nodoc_module'),
(('.. warning:: No documentation found for '
'nodoc in nodoc_module:nodoc'),
'nodoc_module'),
(
(
'.. warning:: No documentation found for '
'nodoc in nodoc_module:nodoc'
),
'nodoc_module',
),
('\n', 'nodoc_module'),
],
results,

40
stevedore/tests/test_test_manager.py
View File

@@ -29,8 +29,12 @@ 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)
a_driver = Extension(
'test_driver',
mock_entry_point,
sentinel.driver_plugin,
sentinel.driver_obj,
)
# base ExtensionManager
@@ -68,8 +72,9 @@ class TestTestManager(utils.TestCase):
func.assert_called_once_with(test_extension)
def test_manager_should_call_all(self):
em = ExtensionManager.make_test_instance([test_extension2,
test_extension])
em = ExtensionManager.make_test_instance(
[test_extension2, test_extension]
)
func = Mock()
em.map(func)
func.assert_any_call(test_extension2)
@@ -79,8 +84,9 @@ class TestTestManager(utils.TestCase):
def mapped(ext, *args, **kwds):
return ext.name
em = ExtensionManager.make_test_instance([test_extension2,
test_extension])
em = ExtensionManager.make_test_instance(
[test_extension2, test_extension]
)
results = em.map(mapped)
self.assertEqual(sorted(results), ['another_one', 'test_extension'])
@@ -93,8 +99,9 @@ class TestTestManager(utils.TestCase):
self.assertEqual(results, [])
def test_manager_should_propagate_exceptions(self):
em = ExtensionManager.make_test_instance([test_extension],
propagate_map_exceptions=True)
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')
@@ -129,7 +136,7 @@ class TestTestManager(utils.TestCase):
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])
assert em[test_extension.name]
def test_hook_manager_should_have_default_namespace(self):
em = HookManager.make_test_instance([test_extension])
@@ -137,8 +144,9 @@ class TestTestManager(utils.TestCase):
def test_hook_manager_should_use_supplied_namespace(self):
namespace = 'testing.1.2.3'
em = HookManager.make_test_instance([test_extension],
namespace=namespace)
em = HookManager.make_test_instance(
[test_extension], namespace=namespace
)
self.assertEqual(namespace, em.namespace)
def test_hook_manager_should_return_named_extensions(self):
@@ -190,8 +198,9 @@ class TestTestManager(utils.TestCase):
self.assertEqual(extensions, em.extensions)
def test_dispatch_map_should_invoke_filter_for_extensions(self):
em = DispatchExtensionManager.make_test_instance([test_extension,
test_extension2])
em = DispatchExtensionManager.make_test_instance(
[test_extension, test_extension2]
)
filter_func = Mock(return_value=False)
args = ('A',)
kw = {'big': 'Cheese'}
@@ -213,8 +222,9 @@ class TestTestManager(utils.TestCase):
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])
em = NameDispatchExtensionManager.make_test_instance(
[test_extension, test_extension2]
)
func = Mock()
args = ('A',)
kw = {'BIGGER': 'Cheese'}

9
tox.ini
View File

@@ -24,11 +24,6 @@ deps =
commands =
pre-commit run -a
[flake8]
ignore = E251,H405
show-source = True
exclude=.venv,.git,.tox,dist,*lib/python*,*egg,build
[testenv:docs]
deps =
-c{env:TOX_CONSTRAINTS_FILE:https://releases.openstack.org/constraints/upper/master}
@@ -50,3 +45,7 @@ deps = bindep
commands = bindep test
usedevelop = False
[flake8]
ignore = E251,H405
show-source = True
exclude = .venv,.tox,dist,doc,releasenotes,*egg,build