Merge branch 'master' into cottsay/superfluous-global-statements

.github/workflows/ci.yaml

@@ -8,7 +8,7 @@ on:
 jobs:
   pytest:
     uses: colcon/ci/.github/workflows/pytest.yaml@main
-    with:
-      codecov: true
+    secrets:
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
   bootstrap:
     uses: ./.github/workflows/bootstrap.yaml

colcon_core/__init__.py

@@ -1,4 +1,4 @@
 # Copyright 2016-2020 Dirk Thomas
 # Licensed under the Apache License, Version 2.0
 
-__version__ = '0.15.2'
+__version__ = '0.16.1'

colcon_core/argument_parser/__init__.py

@@ -42,15 +42,17 @@ def decorate_argument_parser(self, *, parser):
         raise NotImplementedError()
 
 
-def get_argument_parser_extensions():
+def get_argument_parser_extensions(*, group_name=None):
     """
     Get the available argument parser extensions.
 
     The extensions are ordered by their priority and entry point name.
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.ARGUMENT_PARSER_DECORATOR_NAME = name
     return order_extensions_by_priority(extensions)

colcon_core/command.py

@@ -223,20 +223,30 @@ class CustomArgumentParser(argparse.ArgumentParser):
 
         def _parse_optional(self, arg_string):
             result = super()._parse_optional(arg_string)
-            if result == (None, arg_string, None):
+            # Up until https://github.com/python/cpython/pull/114180 ,
+            # _parse_optional() returned a 3-tuple when it couldn't classify
+            # the option.  As of that PR (which is in Python 3.13, and
+            # backported to Python 3.12), it returns a 4-tuple.  Check for
+            # either here.
+            if result in (
+                (None, arg_string, None),
+                (None, arg_string, None, None),
+            ):
                 # in the case there the arg is classified as an unknown 'O'
                 # override that and classify it as an 'A'
                 return None
             return result
 
+    epilog = get_environment_variables_epilog(environment_variables_group_name)
+    if epilog:
+        epilog += '\n\n'
+    epilog += READTHEDOCS_MESSAGE
+
     # top level parser
     parser = CustomArgumentParser(
         prog=get_prog_name(),
         formatter_class=CustomFormatter,
-        epilog=(
-            get_environment_variables_epilog(
-                environment_variables_group_name
-            ) + '\n\n' + READTHEDOCS_MESSAGE))
+        epilog=epilog)
 
     # enable introspecting and intercepting all command line arguments
     parser = decorate_argument_parser(parser)
@@ -284,6 +294,8 @@ def get_environment_variables_epilog(group_name):
     """
     # list environment variables with descriptions
     entry_points = load_extension_points(group_name)
+    if not entry_points:
+        return ''
     env_vars = {
         env_var.name: env_var.description for env_var in entry_points.values()}
     epilog_lines = []
@@ -372,8 +384,6 @@ def create_subparser(parser, cmd_name, verb_extensions, *, attribute):
       selected verb
     :returns: The special action object
     """
-    assert verb_extensions, 'No verb extensions'
-
     # list of available verbs with their descriptions
     verbs = []
     for name, extension in verb_extensions.items():
@@ -383,9 +393,9 @@ def create_subparser(parser, cmd_name, verb_extensions, *, attribute):
     # add subparser with description of verb extensions
     subparser = parser.add_subparsers(
         title=f'{cmd_name} verbs',
-        description='\n'.join(verbs),
+        description='\n'.join(verbs) or None,
         dest=attribute,
-        help=f'call `{cmd_name} VERB -h` for specific help',
+        help=f'call `{cmd_name} VERB -h` for specific help' if verbs else None,
     )
     return subparser
 

colcon_core/environment/__init__.py

@@ -47,15 +47,17 @@ def create_environment_hooks(self, prefix_path, pkg_name):
         raise NotImplementedError()
 
 
-def get_environment_extensions():
+def get_environment_extensions(*, group_name=None):
     """
     Get the available environment extensions.
 
     The extensions are ordered by their priority and entry point name.
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name in list(extensions.keys()):
         extension = extensions[name]
         extension.ENVIRONMENT_NAME = name

colcon_core/event_handler/__init__.py

@@ -44,15 +44,17 @@ def __call__(self, event):
         raise NotImplementedError()
 
 
-def get_event_handler_extensions(*, context):
+def get_event_handler_extensions(*, context, group_name=None):
     """
     Get the available event handler extensions.
 
     The extensions are ordered by their priority and entry point name.
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.EVENT_HANDLER_NAME = name
         extension.context = context

colcon_core/executor/__init__.py

@@ -197,7 +197,7 @@ def _flush(self):
         self._event_controller.flush()
 
 
-def get_executor_extensions():
+def get_executor_extensions(*, group_name=None):
     """
     Get the available executor extensions.
 
@@ -206,7 +206,9 @@ def get_executor_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.EXECUTOR_NAME = name
     return order_extensions_grouped_by_priority(extensions)

colcon_core/extension_point.py

@@ -3,8 +3,11 @@
 # Licensed under the Apache License, Version 2.0
 
 from collections import defaultdict
+from itertools import chain
 import os
+import sys
 import traceback
+import warnings
 
 try:
     from importlib.metadata import distributions
@@ -26,7 +29,6 @@
 
 logger = colcon_logger.getChild(__name__)
 
-
 """
 The group name for entry points identifying colcon extension points.
 
@@ -36,6 +38,8 @@
 """
 EXTENSION_POINT_GROUP_NAME = 'colcon_core.extension_point'
 
+_ENTRY_POINTS_CACHE = []
+
 
 def _get_unique_distributions():
     seen = set()
@@ -46,6 +50,50 @@ def _get_unique_distributions():
             yield dist
 
 
+def _get_entry_points():
+    for dist in _get_unique_distributions():
+        for entry_point in dist.entry_points:
+            # Modern EntryPoint instances should already have this set
+            if not hasattr(entry_point, 'dist'):
+                entry_point.dist = dist
+            yield entry_point
+
+
+def _get_cached_entry_points():
+    if not _ENTRY_POINTS_CACHE:
+        if sys.version_info >= (3, 10):
+            # We prefer using importlib.metadata.entry_points because it
+            # has an internal optimization which allows us to load the entry
+            # points without reading the individual PKG-INFO files, while
+            # still visiting each unique distribution only once.
+            all_entry_points = entry_points()
+            if isinstance(all_entry_points, dict):
+                # Prior to Python 3.12, entry_points returned a (deprecated)
+                # dict. Unfortunately, the "future-proof" recommended
+                # pattern is to add filter parameters, but we actually
+                # want to cache everything so that doesn't work here.
+                with warnings.catch_warnings():
+                    warnings.filterwarnings(
+                        'ignore',
+                        'SelectableGroups dict interface is deprecated',
+                        DeprecationWarning,
+                        module=__name__)
+                    all_entry_points = chain.from_iterable(
+                        all_entry_points.values())
+            _ENTRY_POINTS_CACHE.extend(all_entry_points)
+        else:
+            # If we don't have Python 3.10, we must read each PKG-INFO to
+            # get the name of the distribution so that we can skip the
+            # "shadowed" distributions properly.
+            _ENTRY_POINTS_CACHE.extend(_get_entry_points())
+    return _ENTRY_POINTS_CACHE
+
+
+def clear_entry_point_cache():
+    """Purge the entry point cache."""
+    _ENTRY_POINTS_CACHE.clear()
+
+
 def get_all_extension_points():
     """
     Get all extension points related to `colcon` and any of its extensions.
@@ -59,23 +107,24 @@ def get_all_extension_points():
     colcon_extension_points = get_extension_points(EXTENSION_POINT_GROUP_NAME)
     colcon_extension_points.setdefault(EXTENSION_POINT_GROUP_NAME, None)
 
-    entry_points = defaultdict(dict)
-    for dist in _get_unique_distributions():
-        for entry_point in dist.entry_points:
-            # skip groups which are not registered as extension points
-            if entry_point.group not in colcon_extension_points:
-                continue
-
-            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.metadata['Name'], dist.version)
-    return entry_points
+    extension_points = defaultdict(dict)
+    for entry_point in _get_cached_entry_points():
+        if entry_point.group not in colcon_extension_points:
+            continue
+
+        dist_metadata = entry_point.dist.metadata
+        ep_tuple = (
+            entry_point.value,
+            dist_metadata['Name'], dist_metadata['Version'],
+        )
+        if entry_point.name in extension_points[entry_point.group]:
+            previous = extension_points[entry_point.group][entry_point.name]
+            logger.error(
+                f"Entry point '{entry_point.group}.{entry_point.name}' is "
+                f"declared multiple times, '{ep_tuple}' "
+                f"overwriting '{previous}'")
+        extension_points[entry_point.group][entry_point.name] = ep_tuple
+    return extension_points
 
 
 def get_extension_points(group):
@@ -87,16 +136,9 @@ def get_extension_points(group):
     :rtype: dict
     """
     extension_points = {}
-    try:
-        # Python 3.10 and newer
-        query = entry_points(group=group)
-    except TypeError:
-        query = (
-            entry_point
-            for dist in _get_unique_distributions()
-            for entry_point in dist.entry_points
-            if entry_point.group == group)
-    for entry_point in query:
+    for entry_point in _get_cached_entry_points():
+        if entry_point.group != group:
+            continue
         if entry_point.name in extension_points:
             previous_entry_point = extension_points[entry_point.name]
             logger.error(

colcon_core/logging.py

@@ -95,15 +95,15 @@ def filter(self, record):  # noqa: A003
         if isinstance(handler, logging.StreamHandler):
             formatter = handler.formatter
             # filter colcon specific log messages from default stream handler
-            handler.addFilter(Filter(colcon_logger.name))
+            handler.addFilter(Filter(logger.name))
 
     # add a stream handler replacing the one filtered on the root logger
     handler = logging.StreamHandler()
     if formatter:
         # use same formatter as for stream handler
         handler.setFormatter(formatter)
-    handler.setLevel(colcon_logger.getEffectiveLevel())
-    colcon_logger.addHandler(handler)
+    handler.setLevel(logger.getEffectiveLevel())
+    logger.addHandler(handler)
 
     # add a file handler writing all log levels
     handler = logging.FileHandler(str(path))
@@ -121,9 +121,9 @@ def format_message_with_relative_time(record):
         # use same formatter as for stream handler
         handler.setFormatter(formatter)
     handler.setLevel(1)
-    colcon_logger.addHandler(handler)
+    logger.addHandler(handler)
 
     # change the logger to handle all levels
-    colcon_logger.setLevel(1)
+    logger.setLevel(1)
 
     return handler

colcon_core/package_augmentation/__init__.py

@@ -65,15 +65,17 @@ def augment_package(
         raise NotImplementedError()
 
 
-def get_package_augmentation_extensions():
+def get_package_augmentation_extensions(*, group_name=None):
     """
     Get the available package augmentation extensions.
 
     The extensions are ordered by their priority and entry point name.
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.PACKAGE_AUGMENTATION_NAME = name
     return order_extensions_by_priority(extensions)

colcon_core/package_discovery/__init__.py

@@ -84,15 +84,17 @@ def discover(self, *, args, identification_extensions):
         raise NotImplementedError()
 
 
-def get_package_discovery_extensions():
+def get_package_discovery_extensions(*, group_name=None):
     """
     Get the available package discovery extensions.
 
     The extensions are ordered by their priority and entry point name.
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.PACKAGE_DISCOVERY_NAME = name
     return order_extensions_by_priority(extensions)

colcon_core/package_identification/__init__.py

@@ -64,7 +64,7 @@ def identify(self, desc: PackageDescriptor):
         raise NotImplementedError()
 
 
-def get_package_identification_extensions():
+def get_package_identification_extensions(*, group_name=None):
     """
     Get the available package identification extensions.
 
@@ -73,7 +73,9 @@ def get_package_identification_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.PACKAGE_IDENTIFICATION_NAME = name
     return order_extensions_grouped_by_priority(extensions)