Switch colcon_core.extension_point to importlib.metadata

colcon_core/extension_point.py

@@ -6,11 +6,17 @@
 import os
 import traceback
 
+try:
+    from importlib.metadata import distributions
+    from importlib.metadata import EntryPoint
+    from importlib.metadata import entry_points
+except ImportError:
+    from importlib_metadata import distributions
+    from importlib_metadata import EntryPoint
+    from importlib_metadata import entry_points
+
 from colcon_core.environment_variable import EnvironmentVariable
 from colcon_core.logging import colcon_logger
-from pkg_resources import EntryPoint
-from pkg_resources import iter_entry_points
-from pkg_resources import WorkingSet
 
 """Environment variable to block extensions"""
 EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE = EnvironmentVariable(
@@ -44,27 +50,25 @@ def get_all_extension_points():
     colcon_extension_points.setdefault(EXTENSION_POINT_GROUP_NAME, None)
 
     entry_points = defaultdict(dict)
-    working_set = WorkingSet()
-    for dist in sorted(working_set):
-        entry_map = dist.get_entry_map()
-        for group_name in entry_map.keys():
+    seen = set()
+    for dist in distributions():
+        if dist.name in seen:
+            continue
+        seen.add(dist.name)
+        for entry_point in dist.entry_points:
             # skip groups which are not registered as extension points
-            if group_name not in colcon_extension_points:
+            if entry_point.group not in colcon_extension_points:
                 continue
 
-            group = entry_map[group_name]
-            for entry_point_name, entry_point in group.items():
-                if entry_point_name in entry_points[group_name]:
-                    previous = entry_points[group_name][entry_point_name]
-                    logger.error(
-                        f"Entry point '{group_name}.{entry_point_name}' is "
-                        f"declared multiple times, '{entry_point}' "
-                        f"overwriting '{previous}'")
-                value = entry_point.module_name
-                if entry_point.attrs:
-                    value += f":{'.'.join(entry_point.attrs)}"
-                entry_points[group_name][entry_point_name] = (
-                    value, dist.project_name, getattr(dist, 'version', None))
+            if entry_point.name in entry_points[entry_point.group]:
+                previous = entry_points[entry_point.group][entry_point.name]
+                logger.error(
+                    f"Entry point '{entry_point.group}.{entry_point.name}' is "
+                    f"declared multiple times, '{entry_point.value}' "
+                    f"from '{dist._path}' "
+                    f"overwriting '{previous}'")
+            entry_points[entry_point.group][entry_point.name] = \
+                (entry_point.value, dist.name, dist.version)
     return entry_points
 
 
@@ -76,19 +80,21 @@ def get_extension_points(group):
     :returns: mapping of extension point names to extension point values
     :rtype: dict
     """
-    entry_points = {}
-    for entry_point in iter_entry_points(group=group):
-        if entry_point.name in entry_points:
-            previous_entry_point = entry_points[entry_point.name]
+    extension_points = {}
+    try:
+        # Python 3.10 and newer
+        query = entry_points(group=group)
+    except TypeError:
+        query = entry_points().get(group, ())
+    for entry_point in query:
+        if entry_point.name in extension_points:
+            previous_entry_point = extension_points[entry_point.name]
             logger.error(
                 f"Entry point '{group}.{entry_point.name}' is declared "
-                f"multiple times, '{entry_point}' overwriting "
+                f"multiple times, '{entry_point.value}' overwriting "
                 f"'{previous_entry_point}'")
-        value = entry_point.module_name
-        if entry_point.attrs:
-            value += f":{'.'.join(entry_point.attrs)}"
-        entry_points[entry_point.name] = value
-    return entry_points
+        extension_points[entry_point.name] = entry_point.value
+    return extension_points
 
 
 def load_extension_points(group, *, excludes=None):
@@ -146,10 +152,4 @@ def load_extension_point(name, value, group):
             raise RuntimeError(
                 'The entry point name is listed in the environment variable '
                 f"'{EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE.name}'")
-    if ':' in value:
-        module_name, attr = value.split(':', 1)
-        attrs = attr.split('.')
-    else:
-        module_name = value
-        attrs = ()
-    return EntryPoint(name, module_name, attrs).resolve()
+    return EntryPoint(name, value, group).load()

stdeb.cfg

@@ -1,6 +1,6 @@
 [colcon-core]
 No-Python2:
-Depends3: python3-distlib, python3-empy, python3-pytest, python3-setuptools
+Depends3: python3-distlib, python3-empy, python3-pytest, python3-setuptools, python3 (>= 3.7) | python3-importlib-metadata
 Recommends3: python3-pytest-cov
 Suggests3: python3-pytest-repeat, python3-pytest-rerunfailures
 Suite: bionic focal jammy stretch buster bullseye

test/spell_check.words

@@ -3,7 +3,6 @@ apache
 argcomplete
 argparse
 asyncio
-attrs
 autouse
 basepath
 bazqux

test/test_extension_point.py

@@ -17,96 +17,96 @@
 from .environment_context import EnvironmentContext
 
 
-Group1 = EntryPoint('group1', 'g1')
-Group2 = EntryPoint('group2', 'g2')
+Group1 = EntryPoint('group1', 'g1', EXTENSION_POINT_GROUP_NAME)
+Group2 = EntryPoint('group2', 'g2', EXTENSION_POINT_GROUP_NAME)
+ExtA = EntryPoint('extA', 'eA', Group1.name)
+ExtB = EntryPoint('extB', 'eB', Group1.name)
 
 
 class Dist():
 
-    project_name = 'dist'
+    version = '0.0.0'
 
-    def __init__(self, group_name, group):
-        self._group_name = group_name
-        self._group = group
+    def __init__(self, entry_points):
+        self.name = f'dist-{id(self)}'
+        self._entry_points = entry_points
 
-    def __lt__(self, other):
-        return self._group_name < other._group_name
+    @property
+    def entry_points(self):
+        return list(self._entry_points)
 
-    def get_entry_map(self):
-        return self._group
 
-
-def iter_entry_points(*, group):
+def iter_entry_points(*, group=None):
     if group == EXTENSION_POINT_GROUP_NAME:
         return [Group1, Group2]
-    assert group == Group1.name
-    ep1 = EntryPoint('extA', 'eA')
-    ep2 = EntryPoint('extB', 'eB')
-    return [ep1, ep2]
+    elif group == Group1.name:
+        return [ExtA, ExtB]
+    assert not group
+    return {
+        EXTENSION_POINT_GROUP_NAME: [Group1, Group2],
+        Group1.name: [ExtA, ExtB],
+    }
 
 
-def working_set():
+def distributions():
     return [
-        Dist('group1', {
-            'group1': {ep.name: ep for ep in iter_entry_points(group='group1')}
-        }),
-        Dist('group2', {'group2': {'extC': EntryPoint('extC', 'eC')}}),
-        Dist('groupX', {'groupX': {'extD': EntryPoint('extD', 'eD')}}),
+        Dist(iter_entry_points(group='group1')),
+        Dist([EntryPoint('extC', 'eC', Group2.name)]),
+        Dist([EntryPoint('extD', 'eD', 'groupX')]),
     ]
 
 
 def test_all_extension_points():
     with patch(
-        'colcon_core.extension_point.iter_entry_points',
+        'colcon_core.extension_point.entry_points',
         side_effect=iter_entry_points
     ):
         with patch(
-            'colcon_core.extension_point.WorkingSet',
-            side_effect=working_set
+            'colcon_core.extension_point.distributions',
+            side_effect=distributions
         ):
             # successfully load a known entry point
             extension_points = get_all_extension_points()
             assert set(extension_points.keys()) == {'group1', 'group2'}
             assert set(extension_points['group1'].keys()) == {'extA', 'extB'}
-            assert extension_points['group1']['extA'] == (
-                'eA', Dist.project_name, None)
+            assert extension_points['group1']['extA'][0] == 'eA'
 
 
 def test_extension_point_blocklist():
     # successful loading of extension point without a blocklist
     with patch(
-        'colcon_core.extension_point.iter_entry_points',
+        'colcon_core.extension_point.entry_points',
         side_effect=iter_entry_points
     ):
         with patch(
-            'colcon_core.extension_point.WorkingSet',
-            side_effect=working_set
+            'colcon_core.extension_point.distributions',
+            side_effect=distributions
         ):
             extension_points = get_extension_points('group1')
     assert 'extA' in extension_points.keys()
     extension_point = extension_points['extA']
     assert extension_point == 'eA'
 
-    with patch.object(EntryPoint, 'resolve', return_value=None) as resolve:
+    with patch.object(EntryPoint, 'load', return_value=None) as load:
         load_extension_point('extA', 'eA', 'group1')
-        assert resolve.call_count == 1
+        assert load.call_count == 1
 
         # successful loading of entry point not in blocklist
-        resolve.reset_mock()
+        load.reset_mock()
         with EnvironmentContext(COLCON_EXTENSION_BLOCKLIST=os.pathsep.join([
             'group1.extB', 'group2.extC'])
         ):
             load_extension_point('extA', 'eA', 'group1')
-        assert resolve.call_count == 1
+        assert load.call_count == 1
 
         # entry point in a blocked group can't be loaded
-        resolve.reset_mock()
+        load.reset_mock()
         with EnvironmentContext(COLCON_EXTENSION_BLOCKLIST='group1'):
             with pytest.raises(RuntimeError) as e:
                 load_extension_point('extA', 'eA', 'group1')
             assert 'The entry point group name is listed in the environment ' \
                 'variable' in str(e.value)
-        assert resolve.call_count == 0
+        assert load.call_count == 0
 
         # entry point listed in the blocklist can't be loaded
         with EnvironmentContext(COLCON_EXTENSION_BLOCKLIST=os.pathsep.join([
@@ -116,10 +116,10 @@ def test_extension_point_blocklist():
                 load_extension_point('extA', 'eA', 'group1')
             assert 'The entry point name is listed in the environment ' \
                 'variable' in str(e.value)
-        assert resolve.call_count == 0
+        assert load.call_count == 0
 
 
-def entry_point_resolve(self, *args, **kwargs):
+def entry_point_load(self, *args, **kwargs):
     if self.name == 'exception':
         raise Exception('entry point raising exception')
     if self.name == 'runtime_error':
@@ -129,7 +129,7 @@ def entry_point_resolve(self, *args, **kwargs):
     return DEFAULT
 
 
-@patch.object(EntryPoint, 'resolve', entry_point_resolve)
+@patch.object(EntryPoint, 'load', entry_point_load)
 @patch(
     'colcon_core.extension_point.get_extension_points',
     return_value={'exception': 'a', 'runtime_error': 'b', 'success': 'c'}