Add central interface for defining feature flags

colcon_core/command.py

@@ -52,6 +52,7 @@
 from colcon_core.argument_parser import decorate_argument_parser  # noqa: E402 E501 I100 I202
 from colcon_core.argument_parser import SuppressUsageOutput  # noqa: E402
 from colcon_core.extension_point import load_extension_points  # noqa: E402
+from colcon_core.feature_flags import check_implemented_flags  # noqa: E402
 from colcon_core.location import create_log_path  # noqa: E402
 from colcon_core.location import get_log_path  # noqa: E402
 from colcon_core.location import set_default_config_path  # noqa: E402
@@ -132,6 +133,9 @@ def _main(*, command_name, argv):
         'Command line arguments: {argv}'
         .format(argv=argv if argv is not None else sys.argv))
 
+    # warn about any specified feature flags that are already implemented
+    check_implemented_flags()
+
     # set default locations for config files, for searchability: COLCON_HOME
     set_default_config_path(
         path=(Path('~') / f'.{command_name}').expanduser(),

colcon_core/feature_flags.py

@@ -0,0 +1,71 @@
+# Copyright 2024 Open Source Robotics Foundation, Inc.
+# Licensed under the Apache License, Version 2.0
+
+import os
+
+from colcon_core.environment_variable import EnvironmentVariable
+from colcon_core.logging import colcon_logger
+
+logger = colcon_logger.getChild(__name__)
+
+"""Environment variable to enable feature flags"""
+FEATURE_FLAGS_ENVIRONMENT_VARIABLE = EnvironmentVariable(
+    'COLCON_FEATURE_FLAGS',
+    'Enable pre-production features and behaviors')
+
+_REPORTED_USES = set()
+
+IMPLEMENTED_FLAGS = set()
+
+
+def check_implemented_flags():
+    """Check for and warn about flags which have been implemented."""
+    implemented = IMPLEMENTED_FLAGS.intersection(get_feature_flags())
+    if implemented:
+        logger.warning(
+            'The following feature flags have been implemented and should no '
+            'longer be specified in '
+            f'{FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name}: '
+            f"{','.join(implemented)}")
+
+
+def get_feature_flags():
+    """
+    Retrieve all enabled feature flags.
+
+    :returns: List of enabled flags
+    :rtype: list
+    """
+    return [
+        flag for flag in (
+            os.environ.get(FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name) or ''
+        ).split(os.pathsep) if flag
+    ]
+
+
+def is_feature_flag_set(flag):
+    """
+    Determine if a specific feature flag is enabled.
+
+    Feature flags are case-sensitive and separated by the os-specific path
+    separator character.
+
+    :param str flag: Name of the flag to search for
+
+    :returns: True if the flag is set
+    :rtype: bool
+    """
+    if flag in IMPLEMENTED_FLAGS:
+        return True
+    elif flag in get_feature_flags():
+        if flag not in _REPORTED_USES:
+            if not _REPORTED_USES:
+                logger.warning(
+                    'One or more feature flags have been enabled using the '
+                    f'{FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name} environment '
+                    'variable. These features may be unstable and may change '
+                    'API or behavior at any time.')
+            logger.warning(f'Enabling feature: {flag}')
+            _REPORTED_USES.add(flag)
+        return True
+    return False

test/spell_check.words

@@ -1,3 +1,4 @@
+addfinalizer
 addopts
 apache
 argparse
@@ -35,8 +36,10 @@ docstring
 executables
 exitstatus
 fdopen
+ffoo
 filterwarnings
 foobar
+fooo
 fromhex
 functools
 getcategory
@@ -140,5 +143,6 @@ unittest
 unittests
 unlinking
 unrenamed
+usefixtures
 wildcards
 workaround

test/test_feature_flags.py

@@ -0,0 +1,106 @@
+# Copyright 2024 Open Source Robotics Foundation, Inc.
+# Licensed under the Apache License, Version 2.0
+
+import os
+from unittest.mock import patch
+
+from colcon_core.feature_flags import check_implemented_flags
+from colcon_core.feature_flags import FEATURE_FLAGS_ENVIRONMENT_VARIABLE
+from colcon_core.feature_flags import get_feature_flags
+from colcon_core.feature_flags import is_feature_flag_set
+import pytest
+
+
+_FLAGS_TO_TEST = (
+    ('foo',),
+    ('foo', 'foo'),
+    ('foo', ''),
+    ('', 'foo'),
+    ('', 'foo', ''),
+    ('foo', 'bar'),
+    ('bar', 'foo'),
+    ('bar', 'foo', 'baz'),
+)
+
+
+@pytest.fixture
+def feature_flags_value(request):
+    env = dict(os.environ)
+    if request.param is not None:
+        env[FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name] = os.pathsep.join(
+            request.param)
+    else:
+        env.pop(FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name, None)
+
+    mock_env = patch('colcon_core.feature_flags.os.environ', env)
+    request.addfinalizer(mock_env.stop)
+    mock_env.start()
+    return request.param
+
+
+@pytest.fixture
+def feature_flag_reports(request):
+    reported_uses = patch('colcon_core.feature_flags._REPORTED_USES', set())
+    request.addfinalizer(reported_uses.stop)
+    reported_uses.start()
+    return reported_uses
+
+
+@pytest.mark.parametrize(
+    'feature_flags_value',
+    _FLAGS_TO_TEST,
+    indirect=True)
+@pytest.mark.usefixtures('feature_flags_value', 'feature_flag_reports')
+def test_flag_is_set():
+    with patch('colcon_core.feature_flags.logger.warning') as warn:
+        assert is_feature_flag_set('foo')
+        assert warn.call_count == 2
+        assert is_feature_flag_set('foo')
+        assert warn.call_count == 2
+
+
+@pytest.mark.parametrize(
+    'feature_flags_value',
+    (None, *_FLAGS_TO_TEST),
+    indirect=True)
+@pytest.mark.usefixtures('feature_flags_value', 'feature_flag_reports')
+def test_flag_not_set():
+    with patch('colcon_core.feature_flags.logger.warning') as warn:
+        assert not is_feature_flag_set('')
+        assert not is_feature_flag_set('fo')
+        assert not is_feature_flag_set('oo')
+        assert not is_feature_flag_set('fooo')
+        assert not is_feature_flag_set('ffoo')
+        assert not is_feature_flag_set('qux')
+        assert warn.call_count == 0
+
+
+@pytest.mark.parametrize(
+    'feature_flags_value',
+    (None, *_FLAGS_TO_TEST),
+    indirect=True)
+@pytest.mark.usefixtures('feature_flags_value')
+def test_get_flags(feature_flags_value):
+    assert [
+        flag for flag in (feature_flags_value or ()) if flag
+    ] == get_feature_flags()
+
+
+@pytest.mark.parametrize('feature_flags_value', (('baz',),), indirect=True)
+@pytest.mark.usefixtures('feature_flags_value')
+def test_implemented():
+    with patch('colcon_core.feature_flags.IMPLEMENTED_FLAGS', {'foo'}):
+        with patch('colcon_core.feature_flags.logger.warning') as warn:
+            assert not is_feature_flag_set('bar')
+            assert warn.call_count == 0
+            assert is_feature_flag_set('baz')
+            assert warn.call_count == 2
+            assert is_feature_flag_set('foo')
+            assert warn.call_count == 2
+            check_implemented_flags()
+            assert warn.call_count == 2
+
+    with patch('colcon_core.feature_flags.IMPLEMENTED_FLAGS', {'baz'}):
+        with patch('colcon_core.feature_flags.logger.warning') as warn:
+            check_implemented_flags()
+            assert warn.call_count == 1