Python pkg_resources 模块，EntryPoint() 实例源码

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)

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