Add and use new EnvironDict class

colcon_core/environment_variable.py

@@ -2,7 +2,160 @@
 # Licensed under the Apache License, Version 2.0
 
 from collections import namedtuple
+from collections import OrderedDict
+import sys
 
 
 EnvironmentVariable = namedtuple(
     'EnvironmentVariable', ('name', 'description'))
+
+if sys.platform == 'win32':
+
+    class EnvironDict(OrderedDict):
+        """
+        A case-insensitive dictionary for environment variables.
+
+        This class has to work pretty hard to inherit from
+        :class:`collections.OrderedDict`. In the future, it might be better to
+        document and enforce function return values as implementing
+        :class:`collections.abc.Mapping` rather than actual :class:`dict`
+        instances.
+        """
+
+        def __init__(self, data=None, **kwargs):
+            """Initialize the dictionary."""
+            self._casing = {}
+            super().__init__()
+            if data is None:
+                data = {}
+            self.update(data, **kwargs)
+
+        def __setitem__(self, key, value):
+            """Set value for a key, preserving its original casing."""
+            upper_key = key.upper()
+            if upper_key in self._casing:
+                key = self._casing[upper_key]
+            else:
+                self._casing[upper_key] = key
+            super().__setitem__(key, value)
+
+        def __getitem__(self, key):
+            """Get the value for a key."""
+            upper_key = key.upper()
+            if upper_key in self._casing:
+                orig_key = self._casing[upper_key]
+                try:
+                    return super().__getitem__(orig_key)
+                except KeyError:
+                    del self._casing[upper_key]
+            raise KeyError(key)
+
+        def __delitem__(self, key):
+            """Delete a key."""
+            upper_key = key.upper()
+            if upper_key in self._casing:
+                orig_key = self._casing.pop(upper_key)
+                try:
+                    super().__delitem__(orig_key)
+                except KeyError:
+                    raise KeyError(key)
+            else:
+                raise KeyError(key)
+
+        def __contains__(self, key):
+            """Check if a key exists."""
+            if not isinstance(key, str):
+                return False
+            return key.upper() in self._casing
+
+        def get(self, key, default=None):
+            """Get the value for a key or a default."""
+            if not isinstance(key, str):
+                return default
+            upper_key = key.upper()
+            if upper_key in self._casing:
+                orig_key = self._casing[upper_key]
+                try:
+                    return super().__getitem__(orig_key)
+                except KeyError:
+                    del self._casing[upper_key]
+            return default
+
+        def pop(self, key, *args):
+            """Remove a key and return its value."""
+            upper_key = key.upper()
+            if upper_key in self._casing:
+                orig_key = self._casing.pop(upper_key)
+                try:
+                    return super().pop(orig_key)
+                except KeyError:
+                    pass
+            if args:
+                return args[0]
+            raise KeyError(key)
+
+        def popitem(self, last=True):
+            """Remove and return a (key, value) pair from the dictionary."""
+            key, value = super().popitem(last)
+            self._casing.pop(key.upper(), None)
+            return key, value
+
+        def setdefault(self, key, default=None):
+            """Set default value for a key."""
+            upper_key = key.upper()
+            if upper_key in self._casing:
+                return super().setdefault(self._casing[upper_key], default)
+            self._casing[upper_key] = key
+            return super().setdefault(key, default)
+
+        def update(self, data=None, **kwargs):
+            """Update the dictionary with items from another mapping."""
+            if data is not None:
+                if hasattr(data, 'keys'):
+                    for k in data.keys():
+                        self[k] = data[k]
+                else:
+                    for k, v in data:
+                        self[k] = v
+            for k, v in kwargs.items():
+                self[k] = v
+
+        def clear(self):
+            """Clear the dictionary."""
+            self._casing.clear()
+            super().clear()
+
+        def upper_items(self):
+            """Like iteritems(), but with all uppercase keys."""
+            return (
+                (k.upper(), v)
+                for k, v in self.items()
+            )
+
+        def copy(self):
+            """Return a shallow copy of the dictionary."""
+            return EnvironDict(self)
+
+        def move_to_end(self, key, last=True):
+            """Move an existing element to one end of the dictionary."""
+            upper_key = key.upper()
+            if upper_key in self._casing:
+                super().move_to_end(self._casing[upper_key], last=last)
+            else:
+                raise KeyError(key)
+
+        def __eq__(self, other):
+            """Compare to another mapping case-insensitively."""
+            from collections.abc import Mapping
+            if isinstance(other, Mapping):
+                other = EnvironDict(other)
+            else:
+                return NotImplemented
+            # Compare insensitively
+            return dict(self.upper_items()) == dict(other.upper_items())
+
+        def __repr__(self):
+            """Return the string representation."""
+            return str(dict(self.items()))
+else:
+    EnvironDict = OrderedDict

colcon_core/python_project/hook_caller/__init__.py

@@ -7,6 +7,7 @@
 import os
 import sys
 
+from colcon_core.environment_variable import EnvironDict
 from colcon_core.python_project.hook_caller import _call_hook
 from colcon_core.python_project.hook_caller import _list_hooks
 from colcon_core.python_project.spec import load_and_cache_spec
@@ -59,7 +60,7 @@ def __init__(
         """
         self._backend_name = backend_name
         self._project_path = str(project_path) if project_path else None
-        self._env = dict(env if env is not None else os.environ)
+        self._env = EnvironDict(env if env is not None else os.environ)
         self._stdout_callback = stdout_callback
         self._stderr_callback = stderr_callback
 

colcon_core/shell/__init__.py

@@ -12,6 +12,7 @@
 import warnings
 
 from colcon_core.dependency_descriptor import DependencyDescriptor
+from colcon_core.environment_variable import EnvironDict
 from colcon_core.environment_variable import EnvironmentVariable
 from colcon_core.logging import colcon_logger
 from colcon_core.plugin_system import instantiate_extensions
@@ -346,7 +347,7 @@ async def get_null_separated_environment_variables(
     """
     encoding = locale.getpreferredencoding()
     output = await check_output(cmd, cwd=cwd, shell=shell)
-    env = OrderedDict()
+    env = EnvironDict()
     for kvp in output.split(b'\0'):
         kvp = kvp.rstrip()
         if not kvp:
@@ -377,7 +378,7 @@ async def get_environment_variables(cmd, *, cwd=None, shell=True):
     :rtype: dict
     """
     output = await check_output(cmd, cwd=cwd, shell=shell)
-    env = OrderedDict()
+    env = EnvironDict()
     for line in output.splitlines():
         line = line.rstrip()
         if not line:

colcon_core/task/python/build.py

@@ -11,6 +11,7 @@
 
 from colcon_core.environment import create_environment_hooks
 from colcon_core.environment import create_environment_scripts
+from colcon_core.environment_variable import EnvironDict
 from colcon_core.logging import colcon_logger
 from colcon_core.plugin_system import satisfies_version
 from colcon_core.python_install_path import get_python_install_path
@@ -87,7 +88,7 @@ async def build(self, *, additional_hooks=None):  # noqa: D102
         distutils_commands = os.path.join(
             os.path.dirname(__file__), 'colcon_distutils_commands')
         # and being in the PYTHONPATH
-        env = dict(env)
+        env = EnvironDict(env)
         env['PYTHONPATH'] = str(prefix_override) + os.pathsep + \
             distutils_commands + os.pathsep + \
             python_lib + os.pathsep + env.get('PYTHONPATH', '')

colcon_core/task/python/test/__init__.py

@@ -4,6 +4,7 @@
 import re
 import traceback
 
+from colcon_core.environment_variable import EnvironDict
 from colcon_core.extension_point import load_extension_points
 from colcon_core.logging import colcon_logger
 from colcon_core.package_augmentation.python import extract_dependencies
@@ -74,7 +75,7 @@ async def test(self, *, additional_hooks=None):  # noqa: D102
         logger.log(1, f"test.step() by extension '{key}'")
         try:
             if 'PYTHONDONTWRITEBYTECODE' not in env:
-                env = dict(env)
+                env = EnvironDict(env)
                 env['PYTHONDONTWRITEBYTECODE'] = '1'
             return await extension.step(self.context, env, setup_py_data)
         except Exception as e:  # noqa: F841

colcon_core/task/python/test/pytest.py

@@ -6,6 +6,7 @@
 from pathlib import PurePosixPath
 import sys
 
+from colcon_core.environment_variable import EnvironDict
 from colcon_core.event.test import TestFailure
 from colcon_core.plugin_system import satisfies_version
 from colcon_core.plugin_system import SkipExtensionException
@@ -69,7 +70,7 @@ async def step(self, context, env, setup_py_data):  # noqa: D102
                 '-o', 'cache_dir=' + str(PurePosixPath(
                     *(Path(context.args.build_base).parts)) / '.pytest_cache'),
             ]
-        env = dict(env)
+        env = EnvironDict(env)
 
         if (
             context.args.pytest_with_coverage or

test/test_environment_variable.py

@@ -0,0 +1,94 @@
+# Copyright 2026 Open Source Robotics Foundation, Inc.
+# Licensed under the Apache License, Version 2.0
+
+import sys
+
+from colcon_core.environment_variable import EnvironDict
+from colcon_core.environment_variable import EnvironmentVariable
+import pytest
+
+
+def test_environment_variable():
+    ev = EnvironmentVariable('NAME', 'desc')
+    assert ev.name == 'NAME'
+    assert ev.description == 'desc'
+
+
+def test_environ_dict():
+    if sys.platform == 'win32':
+        return _test_environ_dict_win32()
+    else:
+        return _test_environ_dict_posix()
+
+
+def _test_environ_dict_win32():
+    # Test initialization
+    env = EnvironDict({'Foo': 'bar'})
+    assert len(env) == 1
+
+    # Test case-insensitive get
+    assert env['foo'] == 'bar'
+    assert env['FOO'] == 'bar'
+    assert env['Foo'] == 'bar'
+
+    # Test case-insensitive set and casing preservation
+    env['foo'] = 'baz'
+    assert env['Foo'] == 'baz'
+    assert list(env) == ['Foo']
+
+    # Test kwargs init
+    env2 = EnvironDict(Foo='bar')
+    assert env2['foo'] == 'bar'
+
+    # Test __iter__ yields properly cased keys
+    env3 = EnvironDict()
+    env3['MiXed'] = 'val'
+    assert list(env3) == ['MiXed']
+
+    # Test upper_items
+    assert list(env3.upper_items()) == [('MIXED', 'val')]
+
+    # Test __eq__
+    assert env3 == {'mixed': 'val'}
+    assert env3 == EnvironDict({'MIXED': 'val'})
+    assert env3 != {'mixed': 'other'}
+    assert env3 != 42
+
+    # Test copy
+    env3_copy = env3.copy()
+    assert env3_copy == env3
+    assert list(env3_copy) == ['MiXed']
+
+    # Test delete
+    del env3['mIxEd']
+    assert len(env3) == 0
+    with pytest.raises(KeyError):
+        _ = env3['MiXed']
+
+    # Test __repr__
+    env4 = EnvironDict({'A': '1'})
+    assert repr(env4) == "{'A': '1'}"
+
+    # Test move_to_end
+    env5 = EnvironDict({'A': '1', 'B': '2'})
+    env5.move_to_end('a')
+    assert list(env5) == ['B', 'A']
+    with pytest.raises(KeyError):
+        env5.move_to_end('C')
+
+
+def _test_environ_dict_posix():
+    env = EnvironDict()
+    env['Foo'] = 'bar'
+    assert env['Foo'] == 'bar'
+
+    # On POSIX it should be case-sensitive
+    with pytest.raises(KeyError):
+        _ = env['foo']
+
+    # Test move_to_end is case-sensitive on POSIX
+    env2 = EnvironDict({'Foo': '1', 'Bar': '2'})
+    env2.move_to_end('Foo')
+    assert list(env2) == ['Bar', 'Foo']
+    with pytest.raises(KeyError):
+        env2.move_to_end('foo')

test/test_feature_flags.py

@@ -4,6 +4,7 @@
 import os
 from unittest.mock import patch
 
+from colcon_core.environment_variable import EnvironDict
 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
@@ -25,7 +26,7 @@
 
 @pytest.fixture
 def feature_flags_value(request):
-    env = dict(os.environ)
+    env = EnvironDict(os.environ)
     if request.param is not None:
         env[FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name] = os.pathsep.join(
             request.param)