def test_generate_script(self):
group = 'console_scripts'
entry_point = pkg_resources.EntryPoint(
name='test-ep',
module_name='pbr.packaging',
attrs=('LocalInstallScripts',))
header = '#!/usr/bin/env fake-header\n'
template = ('%(group)s %(module_name)s %(import_target)s '
'%(invoke_target)s')
generated_script = packaging.generate_script(
group, entry_point, header, template)
expected_script = (
'#!/usr/bin/env fake-header\nconsole_scripts pbr.packaging '
'LocalInstallScripts LocalInstallScripts'
)
self.assertEqual(expected_script, generated_script)
python类EntryPoint()的实例源码
def test_generate_script_validates_expectations(self):
group = 'console_scripts'
entry_point = pkg_resources.EntryPoint(
name='test-ep',
module_name='pbr.packaging')
header = '#!/usr/bin/env fake-header\n'
template = ('%(group)s %(module_name)s %(import_target)s '
'%(invoke_target)s')
self.assertRaises(
ValueError, packaging.generate_script, group, entry_point, header,
template)
entry_point = pkg_resources.EntryPoint(
name='test-ep',
module_name='pbr.packaging',
attrs=('attr1', 'attr2', 'attr3'))
self.assertRaises(
ValueError, packaging.generate_script, group, entry_point, header,
template)
def test_adding_entry_points_affect_entry_point_map(easter_fixture):
easter_fixture.stub_egg.add_entry_point_from_line(easter_fixture.group_name,
'test1 = reahl.stubble_dev.test_easteregg:TestClass1')
easter_fixture.stub_egg.add_entry_point(easter_fixture.group_name, 'test2', TestClass2)
epmap = easter_fixture.stub_egg.get_entry_map()
assert list(epmap.keys()) == [easter_fixture.group_name]
name_to_entry_point = list(epmap.values())[0]
assert len(list(name_to_entry_point.keys())) == 2
assert isinstance(name_to_entry_point['test1'], pkg_resources.EntryPoint)
assert name_to_entry_point['test1'].load() is TestClass1
assert isinstance(name_to_entry_point['test2'], pkg_resources.EntryPoint)
assert name_to_entry_point['test2'].load() is TestClass2
easter_fixture.stub_egg.clear()
assert not easter_fixture.stub_egg.get_entry_map()
def _entry_points():
# type: () -> DefaultDict[str, Dict[str, pkg_resources.EntryPoint]]
"""Discover all entry points for required groups if they have not already been found.
:returns: Mapping of group to name to entry points
:rtype: dict
"""
if not _ENTRY_POINTS:
_discover_entry_points()
return _ENTRY_POINTS
def setUp(self):
self.ep1 = pkg_resources.EntryPoint(
"ep1", "p1.ep1", dist=mock.MagicMock(key="p1"))
self.ep1prim = pkg_resources.EntryPoint(
"ep1", "p2.ep2", dist=mock.MagicMock(key="p2"))
# nested
self.ep2 = pkg_resources.EntryPoint(
"ep2", "p2.foo.ep2", dist=mock.MagicMock(key="p2"))
# project name != top-level package name
self.ep3 = pkg_resources.EntryPoint(
"ep3", "a.ep3", dist=mock.MagicMock(key="p3"))
from certbot.plugins.disco import PluginEntryPoint
self.plugin_ep = PluginEntryPoint(EP_SA)
def __init__(self):
# type: () -> None
"""Initialize the data for the configuration."""
self._command = None # type: str
self._subcommands = {} # type: typing.Dict[str, RcliEntryPoint]
self._version = None # type: str
self._entry_point = None # type: pkg_resources.EntryPoint
self._config = {} # type: typing.Dict[str, typing.Any]
if (self.distribution and
self.distribution.has_metadata('rcli-config.json')):
data = self.distribution.get_metadata('rcli-config.json')
self._config = json.loads(data)
def entry_point(self):
# type: () -> pkg_resources.EntryPoint
"""The currently active entry point."""
if not self._entry_point:
for ep in pkg_resources.iter_entry_points(group='console_scripts'):
if (ep.name == self.command and
ep.module_name == self._EP_MOD_NAME):
self._entry_point = ep
return self._entry_point
def setUp(self):
self.ep1 = pkg_resources.EntryPoint(
"ep1", "p1.ep1", dist=mock.MagicMock(key="p1"))
self.ep1prim = pkg_resources.EntryPoint(
"ep1", "p2.ep2", dist=mock.MagicMock(key="p2"))
# nested
self.ep2 = pkg_resources.EntryPoint(
"ep2", "p2.foo.ep2", dist=mock.MagicMock(key="p2"))
# project name != top-level package name
self.ep3 = pkg_resources.EntryPoint(
"ep3", "a.ep3", dist=mock.MagicMock(key="p3"))
from certbot.plugins.disco import PluginEntryPoint
self.plugin_ep = PluginEntryPoint(EP_SA)
def event_entrypoints():
"""Declare some events by mocking the invenio_stats.events entrypoint.
It yields a list like [{event_type: <event_type_name>}, ...].
"""
data = []
result = []
for idx in range(5):
event_type_name = 'event_{}'.format(idx)
from pkg_resources import EntryPoint
entrypoint = EntryPoint(event_type_name, event_type_name)
conf = dict(event_type=event_type_name, templates='/',
processor_class=EventsIndexer)
entrypoint.load = lambda conf=conf: (lambda: [conf])
data.append(entrypoint)
result.append(conf)
# including file_download
event_type_name = 'file-download'
from pkg_resources import EntryPoint
entrypoint = EntryPoint('invenio_files_rest', 'test_dir')
conf = dict(event_type=event_type_name, templates='contrib/file-download',
processor_class=EventsIndexer)
entrypoint.load = lambda conf=conf: (lambda: [conf])
data.append(entrypoint)
entrypoints = mock_iter_entry_points_factory(data, 'invenio_stats.events')
with patch('invenio_stats.ext.iter_entry_points',
entrypoints):
yield result
def add_entry_point_from_line(self, group_name, line):
entry = pkg_resources.EntryPoint.parse(line, dist=self)
self.entry_points[ (group_name, entry.name) ] = entry
return entry
def add_entry_point(self, group_name, name, the_class):
entry = pkg_resources.EntryPoint(name, the_class.__module__,
attrs=(the_class.__name__,),
dist=self)
self.entry_points[ (group_name, entry.name) ] = entry
return entry
def query_entrypoints(custom_permission_factory):
"""Same as event_entrypoints for queries."""
from pkg_resources import EntryPoint
entrypoint = EntryPoint('invenio_stats', 'queries')
data = []
result = []
conf = [dict(
query_name='test-query',
query_class=CustomQuery,
query_config=dict(
index='stats-file-download',
doc_type='file-download-day-aggregation',
copy_fields=dict(
bucket_id='bucket_id',
),
required_filters=dict(
bucket_id='bucket_id',
)
),
permission_factory=custom_permission_factory
),
dict(
query_name='test-query2',
query_class=CustomQuery,
query_config=dict(
index='stats-file-download',
doc_type='file-download-day-aggregation',
copy_fields=dict(
bucket_id='bucket_id',
),
required_filters=dict(
bucket_id='bucket_id',
)
),
permission_factory=custom_permission_factory
)]
result += conf
result += register_queries()
entrypoint.load = lambda conf=conf: (lambda: result)
data.append(entrypoint)
entrypoints = mock_iter_entry_points_factory(data, 'invenio_stats.queries')
with patch('invenio_stats.ext.iter_entry_points',
entrypoints):
yield result
