From c589bee998642b933f182cc64d30416fb4bb1b5f Mon Sep 17 00:00:00 2001 From: Scott K Logan Date: Tue, 18 Aug 2026 14:48:06 -0500 Subject: [PATCH] Add and use new `EnvironDict` class On Win32, the `os.environ` mapping behaves in many case-insensitive ways. When we copy `os.environ` to a typical `dict` instance, we lose those behaviors. This new class preserves the case insensitivity on Windows and behaves like a normal OrderedDict on non-Windows platforms. Assisted-by: Gemini 3.5 Flash --- colcon_core/environment_variable.py | 153 ++++++++++++++++++ .../python_project/hook_caller/__init__.py | 3 +- colcon_core/shell/__init__.py | 5 +- colcon_core/task/python/build.py | 3 +- colcon_core/task/python/test/__init__.py | 3 +- colcon_core/task/python/test/pytest.py | 3 +- test/test_environment_variable.py | 94 +++++++++++ test/test_feature_flags.py | 3 +- 8 files changed, 260 insertions(+), 7 deletions(-) create mode 100644 test/test_environment_variable.py diff --git a/colcon_core/environment_variable.py b/colcon_core/environment_variable.py index 6b1d9069b..d8e4fdc47 100644 --- a/colcon_core/environment_variable.py +++ b/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 diff --git a/colcon_core/python_project/hook_caller/__init__.py b/colcon_core/python_project/hook_caller/__init__.py index 8b285d202..379f44542 100644 --- a/colcon_core/python_project/hook_caller/__init__.py +++ b/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 diff --git a/colcon_core/shell/__init__.py b/colcon_core/shell/__init__.py index a2cd78fca..e85730fca 100644 --- a/colcon_core/shell/__init__.py +++ b/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: diff --git a/colcon_core/task/python/build.py b/colcon_core/task/python/build.py index d200e464a..e7f475391 100644 --- a/colcon_core/task/python/build.py +++ b/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', '') diff --git a/colcon_core/task/python/test/__init__.py b/colcon_core/task/python/test/__init__.py index a01ec4784..46616c44f 100644 --- a/colcon_core/task/python/test/__init__.py +++ b/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 diff --git a/colcon_core/task/python/test/pytest.py b/colcon_core/task/python/test/pytest.py index 335e2140b..b58416768 100644 --- a/colcon_core/task/python/test/pytest.py +++ b/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 diff --git a/test/test_environment_variable.py b/test/test_environment_variable.py new file mode 100644 index 000000000..acaa81cb5 --- /dev/null +++ b/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') diff --git a/test/test_feature_flags.py b/test/test_feature_flags.py index 218b29773..3d65110bc 100644 --- a/test/test_feature_flags.py +++ b/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)