github/nzbToMedia
1
0
Fork 0
mirror of https://github.com/clinton-hall/nzbToMedia.git synced 2025-08-14 18:47:09 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions

Updated stevedore to 2.0.1

Browse source
This commit is contained in:
Labrys of Knossos 2022-11-29 01:47:46 -05:00
commit fb6011f88d
52 changed files with 581 additions and 1960 deletions
Download patch file Download diff file Expand all files Collapse all files

203
libs/common/stevedore/_cache.py
View file

@ -1,203 +0,0 @@
# 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.
"""Use a cache layer in front of entry point scanning."""
import errno
import glob
import hashlib
import itertools
import json
import logging
import os
import os.path
import struct
import sys
try:
# For python 3.8 and later
import importlib.metadata as importlib_metadata
except ImportError:
# For everyone else
import importlib_metadata
log = logging.getLogger('stevedore._cache')
def _get_cache_dir():
"""Locate a platform-appropriate cache directory to use.
Does not ensure that the cache directory exists.
"""
# 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')
return os.path.join(base_path, 'python-entrypoints')
# Mac OS
elif sys.platform == 'darwin':
return os.path.expanduser('~/Library/Caches/Python Entry Points')
# Windows (hopefully)
else:
base_path = os.environ.get('LOCALAPPDATA', None) \
or os.path.expanduser('~\\AppData\\Local')
return os.path.join(base_path, 'Python Entry Points')
def _get_mtime(name):
try:
s = os.stat(name)
return s.st_mtime
except OSError as err:
if err.errno != errno.ENOENT:
raise
return -1.0
def _ftobytes(f):
return struct.Struct('f').pack(f)
def _hash_settings_for_path(path):
"""Return a hash and the path settings that created it."""
paths = []
h = hashlib.sha256()
# Tie the cache to the python interpreter, in case it is part of a
# virtualenv.
h.update(sys.executable.encode('utf-8'))
h.update(sys.prefix.encode('utf-8'))
for entry in path:
mtime = _get_mtime(entry)
h.update(entry.encode('utf-8'))
h.update(_ftobytes(mtime))
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'))
):
mtime = _get_mtime(ep_file)
h.update(ep_file.encode('utf-8'))
h.update(_ftobytes(mtime))
paths.append((ep_file, mtime))
return (h.hexdigest(), paths)
def _build_cacheable_data(path):
real_groups = importlib_metadata.entry_points()
# Convert the namedtuple values to regular tuples
groups = {}
for name, group_data in real_groups.items():
existing = set()
members = []
groups[name] = members
for ep in group_data:
# Filter out duplicates that can occur when testing a
# package that provides entry points using tox, where the
# package is installed in the virtualenv that tox builds
# and is present in the path as '.'.
item = ep.name, ep.value, ep.group # convert to tuple
if item in existing:
continue
existing.add(item)
members.append(item)
return {
'groups': groups,
'sys.executable': sys.executable,
'sys.prefix': sys.prefix,
}
class Cache:
def __init__(self, cache_dir=None):
if cache_dir is None:
cache_dir = _get_cache_dir()
self._dir = cache_dir
self._internal = {}
self._disable_caching = False
# 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']):
self._disable_caching = True
def _get_data_for_path(self, path):
if path is None:
path = sys.path
internal_key = tuple(path)
if internal_key in self._internal:
return self._internal[internal_key]
digest, path_values = _hash_settings_for_path(path)
filename = os.path.join(self._dir, digest)
try:
log.debug('reading %s', filename)
with open(filename, 'r') as f:
data = json.load(f)
except (IOError, json.JSONDecodeError):
data = _build_cacheable_data(path)
data['path_values'] = path_values
if not self._disable_caching:
try:
log.debug('writing to %s', filename)
os.makedirs(self._dir, exist_ok=True)
with open(filename, 'w') as f:
json.dump(data, f)
except (IOError, OSError):
# Could not create cache dir or write file.
pass
self._internal[internal_key] = data
return data
def get_group_all(self, group, path=None):
result = []
data = self._get_data_for_path(path)
group_data = data.get('groups', {}).get(group, [])
for vals in group_data:
result.append(importlib_metadata.EntryPoint(*vals))
return result
def get_group_named(self, group, path=None):
result = {}
for ep in self.get_group_all(group, path=path):
if ep.name not in result:
result[ep.name] = ep
return result
def get_single(self, group, name, path=None):
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))
_c = Cache()
get_group_all = _c.get_group_all
get_group_named = _c.get_group_named
get_single = _c.get_single

2
libs/common/stevedore/dispatch.py
View file

@ -142,7 +142,7 @@ class NameDispatchExtensionManager(DispatchExtensionManager):
then ignored
:type invoke_on_load: bool
:param on_load_failure_callback: Callback function that will be called when
an entrypoint can not be loaded. The arguments that will be provided
a entrypoint can not be loaded. The arguments that will be provided
when this is called (when an entrypoint fails to load) are
(manager, entrypoint, exception)
:type on_load_failure_callback: function

10
libs/common/stevedore/driver.py
View file

@ -10,8 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from .exception import MultipleMatches
from .exception import NoMatches
from .exception import NoMatches, MultipleMatches
from .named import NamedExtensionManager
@ -34,7 +33,7 @@ class DriverManager(NamedExtensionManager):
is True.
:type invoke_kwds: dict
:param on_load_failure_callback: Callback function that will be called when
an entrypoint can not be loaded. The arguments that will be provided
a entrypoint can not be loaded. The arguments that will be provided
when this is called (when an entrypoint fails to load) are
(manager, entrypoint, exception)
:type on_load_failure_callback: function
@ -86,7 +85,7 @@ class DriverManager(NamedExtensionManager):
and then ignored
:type propagate_map_exceptions: bool
:param on_load_failure_callback: Callback function that will
be called when an entrypoint can not be loaded. The
be called when a entrypoint can not be loaded. The
arguments that will be provided when this is called (when
an entrypoint fails to load) are (manager, entrypoint,
exception)
@ -143,6 +142,7 @@ class DriverManager(NamedExtensionManager):
@property
def driver(self):
"""Returns the driver being used by this manager."""
"""Returns the driver being used by this manager.
"""
ext = self.extensions[0]
return ext.obj if ext.obj else ext.plugin

2
libs/common/stevedore/enabled.py
View file

@ -46,7 +46,7 @@ class EnabledExtensionManager(ExtensionManager):
then ignored
:type propagate_map_exceptions: bool
:param on_load_failure_callback: Callback function that will be called when
an entrypoint can not be loaded. The arguments that will be provided
a entrypoint can not be loaded. The arguments that will be provided
when this is called (when an entrypoint fails to load) are
(manager, entrypoint, exception)
:type on_load_failure_callback: function

15
libs/common/stevedore/example/base.py
View file

@ -1,18 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2020 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import abc

14
libs/common/stevedore/example/load_as_driver.py
View file

@ -1,17 +1,3 @@
# Copyright (C) 2020 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
from stevedore import driver

14
libs/common/stevedore/example/load_as_extension.py
View file

@ -1,17 +1,3 @@
# Copyright (C) 2020 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
from stevedore import extension

17
libs/common/stevedore/example/setup.py
View file

@ -1,19 +1,4 @@
# Copyright (C) 2020 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from setuptools import find_packages
from setuptools import setup
from setuptools import setup, find_packages
setup(
name='stevedore-examples',

16
libs/common/stevedore/example/simple.py
View file

@ -1,21 +1,9 @@
# Copyright (C) 2020 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from stevedore.example import base
class Simple(base.FormatterBase):
"""A very basic formatter."""
"""A very basic formatter.
"""
def format(self, data):
"""Format the data and return unicode text.

15
libs/common/stevedore/example2/fields.py
View file

@ -1,18 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2020 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import textwrap
from stevedore.example import base

17
libs/common/stevedore/example2/setup.py
View file

@ -1,19 +1,4 @@
# Copyright (C) 2020 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from setuptools import find_packages
from setuptools import setup
from setuptools import setup, find_packages
setup(
name='stevedore-examples2',

56
libs/common/stevedore/extension.py
View file

@ -13,10 +13,11 @@
"""ExtensionManager
"""
import logging
import operator
import pkg_resources
import logging
from . import _cache
from .exception import NoMatches
LOG = logging.getLogger(__name__)
@ -33,7 +34,7 @@ class Extension(object):
:param name: The entry point name.
:type name: str
:param entry_point: The EntryPoint instance returned by
:mod:`entrypoints`.
:mod:`pkg_resources`.
:type entry_point: EntryPoint
:param plugin: The value returned by entry_point.load()
:param obj: The object returned by ``plugin(*args, **kwds)`` if the
@ -47,38 +48,6 @@ class Extension(object):
self.plugin = plugin
self.obj = obj
@property
def module_name(self):
"""The name of the module from which the entry point is loaded.
:return: A string in 'dotted.module' format.
"""
# NOTE: importlib_metadata from PyPI includes this but the
# Python 3.8 standard library does not.
match = self.entry_point.pattern.match(self.entry_point.value)
return match.group('module')
@property
def extras(self):
"""The 'extras' settings for the plugin."""
# NOTE: The underlying package returns re.Match objects for
# some reason. Translate those to the matched strings, which
# seem more useful.
return [
# Python 3.6 returns _sre.SRE_Match objects. Later
# versions of python return re.Match objects. Both types
# have a 'string' attribute containing the text that
# matched the pattern.
getattr(e, 'string', e)
for e in self.entry_point.extras
]
@property
def attr(self):
"""The attribute of the module to be loaded."""
match = self.entry_point.pattern.match(self.entry_point.value)
return match.group('attr')
@property
def entry_point_target(self):
"""The module and attribute referenced by this extension's entry_point.
@ -86,7 +55,8 @@ class Extension(object):
:return: A string representation of the target of the entry point in
'dotted.module:object' format.
"""
return self.entry_point.value
return '%s:%s' % (self.entry_point.module_name,
self.entry_point.attrs[0])
class ExtensionManager(object):
@ -110,7 +80,7 @@ class ExtensionManager(object):
then ignored
:type propagate_map_exceptions: bool
:param on_load_failure_callback: Callback function that will be called when
an entrypoint can not be loaded. The arguments that will be provided
a entrypoint can not be loaded. The arguments that will be provided
when this is called (when an entrypoint fails to load) are
(manager, entrypoint, exception)
:type on_load_failure_callback: function
@ -156,7 +126,7 @@ class ExtensionManager(object):
are logged and then ignored
:type propagate_map_exceptions: bool
:param on_load_failure_callback: Callback function that will
be called when an entrypoint can not be loaded. The
be called when a entrypoint can not be loaded. The
arguments that will be provided when this is called (when
an entrypoint fails to load) are (manager, entrypoint,
exception)
@ -204,7 +174,7 @@ class ExtensionManager(object):
"""
if self.namespace not in self.ENTRY_POINT_CACHE:
eps = list(_cache.get_group_all(self.namespace))
eps = list(pkg_resources.iter_entry_points(self.namespace))
self.ENTRY_POINT_CACHE[self.namespace] = eps
return self.ENTRY_POINT_CACHE[self.namespace]
@ -252,7 +222,7 @@ class ExtensionManager(object):
ep.require()
plugin = ep.resolve()
else:
plugin = ep.load()
plugin = ep.load(require=verify_requirements)
if invoke_on_load:
obj = plugin(*invoke_args, **invoke_kwds)
else:
@ -331,7 +301,8 @@ class ExtensionManager(object):
LOG.exception(err)
def items(self):
"""Return an iterator of tuples of the form (name, extension).
"""
Return an iterator of tuples of the form (name, extension).
This is analogous to the Mapping.items() method.
"""
@ -355,5 +326,6 @@ class ExtensionManager(object):
return self._extensions_by_name[name]
def __contains__(self, name):
"""Return true if name is in list of enabled extensions."""
"""Return true if name is in list of enabled extensions.
"""
return any(extension.name == name for extension in self.extensions)

2
libs/common/stevedore/hook.py
View file

@ -32,7 +32,7 @@ class HookManager(NamedExtensionManager):
is True.
:type invoke_kwds: dict
:param on_load_failure_callback: Callback function that will be called when
an entrypoint can not be loaded. The arguments that will be provided
a entrypoint can not be loaded. The arguments that will be provided
when this is called (when an entrypoint fails to load) are
(manager, entrypoint, exception)
:type on_load_failure_callback: function

4
libs/common/stevedore/named.py
View file

@ -46,7 +46,7 @@ class NamedExtensionManager(ExtensionManager):
then ignored
:type propagate_map_exceptions: bool
:param on_load_failure_callback: Callback function that will be called when
an entrypoint can not be loaded. The arguments that will be provided
a entrypoint can not be loaded. The arguments that will be provided
when this is called (when an entrypoint fails to load) are
(manager, entrypoint, exception)
:type on_load_failure_callback: function
@ -108,7 +108,7 @@ class NamedExtensionManager(ExtensionManager):
and then ignored
:type propagate_map_exceptions: bool
:param on_load_failure_callback: Callback function that will
be called when an entrypoint can not be loaded. The
be called when a entrypoint can not be loaded. The
arguments that will be provided when this is called (when
an entrypoint fails to load) are (manager, entrypoint,
exception)

27
libs/common/stevedore/sphinxext.py
View file

@ -34,32 +34,29 @@ def _simple_list(mgr):
doc = _get_docstring(ext.plugin) or '\n'
summary = doc.splitlines()[0].strip()
yield('* %s -- %s' % (ext.name, summary),
ext.module_name)
ext.entry_point.module_name)
def _detailed_list(mgr, over='', under='-', titlecase=False):
for name in sorted(mgr.names()):
ext = mgr[name]
if over:
yield (over * len(ext.name), ext.module_name)
yield (over * len(ext.name), ext.entry_point.module_name)
if titlecase:
yield (ext.name.title(), ext.module_name)
yield (ext.name.title(), ext.entry_point.module_name)
else:
yield (ext.name, ext.module_name)
yield (ext.name, ext.entry_point.module_name)
if under:
yield (under * len(ext.name), ext.module_name)
yield ('\n', ext.module_name)
yield (under * len(ext.name), ext.entry_point.module_name)
yield ('\n', ext.entry_point.module_name)
doc = _get_docstring(ext.plugin)
if doc:
yield (doc, ext.module_name)
yield (doc, ext.entry_point.module_name)
else:
yield (
'.. warning:: No documentation found for {} in {}'.format(
ext.name, ext.entry_point_target,
),
ext.module_name,
)
yield ('\n', ext.module_name)
yield ('.. warning:: No documentation found in %s'
% ext.entry_point,
ext.entry_point.module_name)
yield ('\n', ext.entry_point.module_name)
class ListPluginsDirective(rst.Directive):
@ -82,7 +79,7 @@ class ListPluginsDirective(rst.Directive):
underline_style = self.options.get('underline-style', '=')
def report_load_failure(mgr, ep, err):
LOG.warning(u'Failed to load %s: %s' % (ep.module, err))
LOG.warning(u'Failed to load %s: %s' % (ep.module_name, err))
mgr = extension.ExtensionManager(
namespace,

56
libs/common/stevedore/tests/test_cache.py
View file

@ -1,56 +0,0 @@
# 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._cache
"""
import sys
from unittest import mock
from stevedore import _cache
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)
"""
with mock.patch.object(sys, 'executable', '/tmp/fake'):
sot = _cache.Cache()
self.assertTrue(sot._disable_caching)
def test_disable_caching_file(self):
"""Test caching is disabled if .disable file is present in target
dir
"""
cache_dir = _cache._get_cache_dir()
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)
self.assertTrue(sot._disable_caching)
mock_path.return_value = False
sot = _cache.Cache()
self.assertFalse(sot._disable_caching)
@mock.patch('os.makedirs')
@mock.patch('builtins.open')
def test__get_data_for_path_no_write(self, mock_open, mock_mkdir):
sot = _cache.Cache()
sot._disable_caching = True
mock_open.side_effect = IOError
sot._get_data_for_path('fake')
mock_mkdir.assert_not_called()

2
libs/common/stevedore/tests/test_dispatch.py
View file

@ -10,8 +10,8 @@
# License for the specific language governing permissions and limitations
# under the License.
from stevedore import dispatch
from stevedore.tests import utils
from stevedore import dispatch
def check_dispatch(ep, *args, **kwds):

13
libs/common/stevedore/tests/test_driver.py
View file

@ -13,12 +13,7 @@
"""Tests for stevedore.extension
"""
try:
# For python 3.8 and later
import importlib.metadata as importlib_metadata
except ImportError:
# For everyone else
import importlib_metadata
import pkg_resources
from stevedore import driver
from stevedore import exception
@ -73,15 +68,13 @@ class TestCallback(utils.TestCase):
extensions = [
extension.Extension(
'backend',
importlib_metadata.EntryPoint(
'backend', 'pkg1:driver', 'backend'),
pkg_resources.EntryPoint.parse('backend = pkg1:driver'),
'pkg backend',
None,
),
extension.Extension(
'backend',
importlib_metadata.EntryPoint(
'backend', 'pkg2:driver', 'backend'),
pkg_resources.EntryPoint.parse('backend = pkg2:driver'),
'pkg backend',
None,
),

56
libs/common/stevedore/tests/test_extension.py
View file

@ -16,13 +16,6 @@
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
@ -103,13 +96,13 @@ class TestCallback(utils.TestCase):
def test_use_cache(self):
# If we insert something into the cache of entry points,
# the manager should not have to call into entrypoints
# 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('stevedore._cache.get_group_all',
with mock.patch('pkg_resources.iter_entry_points',
side_effect=
AssertionError('called get_group_all')):
AssertionError('called iter_entry_points')):
em = extension.ExtensionManager('stevedore.test.faux')
names = em.names()
self.assertEqual(names, [])
@ -242,48 +235,9 @@ class TestLoadRequirementsOldSetuptools(utils.TestCase):
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()
self.mock_ep.load.assert_called_once_with(require=True)
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)
self.mock_ep.load.assert_called_once_with(require=False)

19
libs/common/stevedore/tests/test_sphinxext.py
View file

@ -12,26 +12,24 @@
"""Tests for the sphinx extension
"""
try:
# For python 3.8 and later
import importlib.metadata as importlib_metadata
except ImportError:
# For everyone else
import importlib_metadata
from unittest import mock
from stevedore import extension
from stevedore import sphinxext
from stevedore.tests import utils
import pkg_resources
def _make_ext(name, docstring):
def inner():
pass
inner.__doc__ = docstring
m1 = importlib_metadata.EntryPoint(
name, '{}_module:{}'.format(name, name), 'group',
)
m1 = mock.Mock(spec=pkg_resources.EntryPoint)
m1.module_name = '%s_module' % name
s = mock.Mock(return_value='ENTRY_POINT(%s)' % name)
m1.__str__ = s
return extension.Extension(name, m1, inner, None)
@ -113,8 +111,7 @@ class TestSphinxExt(utils.TestCase):
('nodoc', 'nodoc_module'),
('-----', 'nodoc_module'),
('\n', 'nodoc_module'),
(('.. warning:: No documentation found for '
'nodoc in nodoc_module:nodoc'),
('.. warning:: No documentation found in ENTRY_POINT(nodoc)',
'nodoc_module'),
('\n', 'nodoc_module'),
],

15
libs/common/stevedore/tests/test_test_manager.py
View file

@ -10,20 +10,15 @@
# License for the specific language governing permissions and limitations
# under the License.
from unittest.mock import Mock
from unittest.mock import sentinel
from unittest.mock import Mock, sentinel
from stevedore.dispatch import DispatchExtensionManager
from stevedore.dispatch import NameDispatchExtensionManager
from stevedore import (ExtensionManager, NamedExtensionManager, HookManager,
DriverManager, EnabledExtensionManager)
from stevedore.dispatch import (DispatchExtensionManager,
NameDispatchExtensionManager)
from stevedore.extension import Extension
from stevedore.tests import utils
from stevedore import DriverManager
from stevedore import EnabledExtensionManager
from stevedore import ExtensionManager
from stevedore import HookManager
from stevedore import NamedExtensionManager
test_extension = Extension('test_extension', None, None, None)
test_extension2 = Extension('another_one', None, None, None)