From e31f37fd5bef8cbe98e60a7ae8ef0a2561b1c162 Mon Sep 17 00:00:00 2001 From: Scott K Logan Date: Tue, 21 Oct 2025 13:40:00 -0500 Subject: [PATCH] Enhance tests for 'sh' and 'bat' shells The existing tests exercise much of the code for implementing colcon's shell subsystem, but it doesn't validate the results. If the platform supports the shell, we should check that the code for appending and prepending paths to lists function as intended. --- colcon_core/shell/template/prefix_util.py.em | 2 +- test/test_shell_bat.py | 188 +++++++++++++++-- test/test_shell_sh.py | 205 ++++++++++++++++--- 3 files changed, 352 insertions(+), 43 deletions(-) diff --git a/colcon_core/shell/template/prefix_util.py.em b/colcon_core/shell/template/prefix_util.py.em index 9ed25d2fd..2bc515add 100644 --- a/colcon_core/shell/template/prefix_util.py.em +++ b/colcon_core/shell/template/prefix_util.py.em @@ -377,7 +377,7 @@ def _remove_ending_separators(): commands = [] for name in env_state: # skip variables that already had values before this script started prepending - if name in os.environ: + if os.environ.get(name): continue commands += [ FORMAT_STR_REMOVE_LEADING_SEPARATOR.format_map({'name': name}), diff --git a/test/test_shell_bat.py b/test/test_shell_bat.py index 29a972ab1..b774a9e5d 100644 --- a/test/test_shell_bat.py +++ b/test/test_shell_bat.py @@ -5,10 +5,14 @@ from pathlib import Path import sys from tempfile import TemporaryDirectory +from unittest.mock import patch from colcon_core import shell +from colcon_core.location import get_relative_package_index_path from colcon_core.plugin_system import SkipExtensionException +from colcon_core.shell import get_environment_variables from colcon_core.shell.bat import BatShell +from colcon_core.shell.dsv import DsvShell import pytest from .run_until_complete import run_until_complete @@ -20,6 +24,7 @@ def test_extension(): try: with TemporaryDirectory(prefix='test_colcon_') as prefix_path: _test_extension(Path(prefix_path)) + _test_prefix_script(Path(prefix_path)) finally: shell.use_all_shell_extensions = use_all_shell_extensions @@ -39,33 +44,36 @@ def _test_extension(prefix_path): extension.create_prefix_script(prefix_path, False) assert (prefix_path / 'local_setup.bat').exists() + # create_hook_append_value + append_hook_path = extension.create_hook_append_value( + 'append_env_hook_name', prefix_path, 'pkg_name', + 'APPEND_NAME', 'subdirectory') + assert append_hook_path.exists() + assert append_hook_path.name == 'append_env_hook_name.bat' + content = append_hook_path.read_text() + assert 'APPEND_NAME' in content + + # create_hook_prepend_value + prepend_hook_path = extension.create_hook_prepend_value( + 'prepend_env_hook_name', prefix_path, 'pkg_name', + 'PREPEND_NAME', 'subdirectory') + assert prepend_hook_path.exists() + assert prepend_hook_path.name == 'prepend_env_hook_name.bat' + content = prepend_hook_path.read_text() + assert 'PREPEND_NAME' in content + # create_package_script extension.create_package_script( prefix_path, 'pkg_name', [ - ('hookA.bat', '/some/path/hookA.bat'), + (append_hook_path.relative_to(prefix_path), ()), + (prepend_hook_path.relative_to(prefix_path), ()), ('hookB.other', '/some/path/hookB.other')]) assert (prefix_path / 'share' / 'pkg_name' / 'package.bat').exists() content = (prefix_path / 'share' / 'pkg_name' / 'package.bat').read_text() - assert 'hookA' in content + assert append_hook_path.name in content + assert prepend_hook_path.name in content assert 'hookB' not in content - # create_hook_append_value - hook_path = extension.create_hook_append_value( - 'append_env_hook_name', prefix_path, 'pkg_name', - 'APPEND_NAME', 'append_subdirectory') - assert hook_path.exists() - assert hook_path.name == 'append_env_hook_name.bat' - content = hook_path.read_text() - assert 'APPEND_NAME' in content - - # create_hook_prepend_value - hook_path = extension.create_hook_prepend_value( - 'env_hook_name', prefix_path, 'pkg_name', 'NAME', 'subdirectory') - assert hook_path.exists() - assert hook_path.name == 'env_hook_name.bat' - content = hook_path.read_text() - assert 'NAME' in content - # generate_command_environment if sys.platform != 'win32': with pytest.raises(SkipExtensionException) as e: @@ -93,3 +101,145 @@ def _test_extension(prefix_path): 'task_name', prefix_path, {'dep': str(prefix_path)}) env = run_until_complete(coroutine) assert isinstance(env, dict) + + subdirectory_path = str(prefix_path / 'subdirectory') + + # validate appending/prepending without existing values + with patch.dict(os.environ) as env_patch: + env_patch.pop('APPEND_NAME', None) + env_patch.pop('PREPEND_NAME', None) + coroutine = extension.generate_command_environment( + 'task_name', prefix_path, {'pkg_name': str(prefix_path)}) + env = run_until_complete(coroutine) + assert env.get('APPEND_NAME') == subdirectory_path + assert env.get('PREPEND_NAME') == subdirectory_path + + # validate appending/prepending with existing values + with patch.dict(os.environ, { + 'APPEND_NAME': 'control', + 'PREPEND_NAME': 'control', + }): + coroutine = extension.generate_command_environment( + 'task_name', prefix_path, {'pkg_name': str(prefix_path)}) + env = run_until_complete(coroutine) + assert env.get('APPEND_NAME') == os.pathsep.join(( + 'control', + subdirectory_path, + )) + assert env.get('PREPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + + # validate appending/prepending unique values + with patch.dict(os.environ, { + 'APPEND_NAME': os.pathsep.join((subdirectory_path, 'control')), + 'PREPEND_NAME': os.pathsep.join(('control', subdirectory_path)), + }): + coroutine = extension.generate_command_environment( + 'task_name', prefix_path, {'pkg_name': str(prefix_path)}) + env = run_until_complete(coroutine) + # Expect no change, value already appears earlier in the list + assert env.get('APPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + # Expect value to be *moved* to the front of the list + assert env.get('PREPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + + +def _test_prefix_script(prefix_path): + extension = BatShell() + + # BatShell requires the (non-primary) DsvShell extension as well + dsv_extension = DsvShell() + + # create_hook_append_value + append_hook_path = dsv_extension.create_hook_append_value( + 'append_env_hook_name', prefix_path, 'pkg_name', + 'APPEND_NAME', 'subdirectory') + assert append_hook_path.exists() + assert append_hook_path.name == 'append_env_hook_name.dsv' + content = append_hook_path.read_text() + assert 'APPEND_NAME' in content + + # create_hook_prepend_value + prepend_hook_path = dsv_extension.create_hook_prepend_value( + 'prepend_env_hook_name', prefix_path, 'pkg_name', + 'PREPEND_NAME', 'subdirectory') + assert prepend_hook_path.exists() + assert prepend_hook_path.name == 'prepend_env_hook_name.dsv' + content = prepend_hook_path.read_text() + assert 'PREPEND_NAME' in content + + # mark package as installed + package_index = prefix_path / get_relative_package_index_path() + package_index.mkdir(parents=True) + (package_index / 'pkg_name').write_text('') + + # create_package_script + dsv_extension.create_package_script( + prefix_path, 'pkg_name', [ + (append_hook_path.relative_to(prefix_path), ()), + (prepend_hook_path.relative_to(prefix_path), ())]) + assert (prefix_path / 'share' / 'pkg_name' / 'package.dsv').exists() + content = (prefix_path / 'share' / 'pkg_name' / 'package.dsv').read_text() + assert append_hook_path.name in content + assert prepend_hook_path.name in content + + # create_prefix_script + extension.create_prefix_script(prefix_path, True) + prefix_script = prefix_path / 'local_setup.bat' + prefix_script.exists() + assert (prefix_path / '_local_setup_util_bat.py').exists() + + if sys.platform == 'win32': + subdirectory_path = str(prefix_path / 'subdirectory') + + # validate appending/prepending without existing values + with patch.dict(os.environ, { + 'APPEND_NAME': '', + 'PREPEND_NAME': '', + }): + coroutine = get_environment_variables([prefix_script, '&&', 'set']) + env = run_until_complete(coroutine) + assert env.get('APPEND_NAME') == subdirectory_path + assert env.get('PREPEND_NAME') == subdirectory_path + + # validate appending/prepending with existing values + with patch.dict(os.environ, { + 'APPEND_NAME': 'control', + 'PREPEND_NAME': 'control', + }): + coroutine = get_environment_variables([prefix_script, '&&', 'set']) + env = run_until_complete(coroutine) + assert env.get('APPEND_NAME') == os.pathsep.join(( + 'control', + subdirectory_path, + )) + assert env.get('PREPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + + # validate appending/prepending unique values + with patch.dict(os.environ, { + 'APPEND_NAME': os.pathsep.join((subdirectory_path, 'control')), + 'PREPEND_NAME': os.pathsep.join(('control', subdirectory_path)), + }): + coroutine = get_environment_variables([prefix_script, '&&', 'set']) + env = run_until_complete(coroutine) + # Expect no change, value already appears earlier in the list + assert env.get('APPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + # TODO: The DsvShell behavior doesn't align with BatShell! + # ~~Expect value to be *moved* to the front of the list~~ + assert env.get('PREPEND_NAME') == os.pathsep.join(( + 'control', + subdirectory_path, + )) diff --git a/test/test_shell_sh.py b/test/test_shell_sh.py index 26b3c4b95..11a1c1122 100644 --- a/test/test_shell_sh.py +++ b/test/test_shell_sh.py @@ -5,9 +5,14 @@ from pathlib import Path import sys from tempfile import TemporaryDirectory +from unittest.mock import patch from colcon_core import shell +from colcon_core.location import get_relative_package_index_path from colcon_core.plugin_system import SkipExtensionException +from colcon_core.shell import get_environment_variables +from colcon_core.shell import get_null_separated_environment_variables +from colcon_core.shell.dsv import DsvShell from colcon_core.shell.sh import ShShell import pytest @@ -20,6 +25,7 @@ def test_extension(): try: with TemporaryDirectory(prefix='test_colcon_') as prefix_path: _test_extension(Path(prefix_path)) + _test_prefix_script(Path(prefix_path)) finally: shell.use_all_shell_extensions = use_all_shell_extensions @@ -35,38 +41,36 @@ def test_extension(): def _test_extension(prefix_path): extension = ShShell() - # create_prefix_script - extension.create_prefix_script(prefix_path, False) - assert (prefix_path / 'local_setup.sh').exists() - assert (prefix_path / '_local_setup_util_sh.py').exists() + # create_hook_append_value + append_hook_path = extension.create_hook_append_value( + 'append_env_hook_name', prefix_path, 'pkg_name', + 'APPEND_NAME', 'subdirectory') + assert append_hook_path.exists() + assert append_hook_path.name == 'append_env_hook_name.sh' + content = append_hook_path.read_text() + assert 'APPEND_NAME' in content + + # create_hook_prepend_value + prepend_hook_path = extension.create_hook_prepend_value( + 'prepend_env_hook_name', prefix_path, 'pkg_name', + 'PREPEND_NAME', 'subdirectory') + assert prepend_hook_path.exists() + assert prepend_hook_path.name == 'prepend_env_hook_name.sh' + content = prepend_hook_path.read_text() + assert 'PREPEND_NAME' in content # create_package_script extension.create_package_script( prefix_path, 'pkg_name', [ - ('hookA.sh', '/some/path/hookA.sh'), + (append_hook_path.relative_to(prefix_path), ()), + (prepend_hook_path.relative_to(prefix_path), ()), ('hookB.other', '/some/path/hookB.other')]) assert (prefix_path / 'share' / 'pkg_name' / 'package.sh').exists() content = (prefix_path / 'share' / 'pkg_name' / 'package.sh').read_text() - assert 'hookA' in content + assert append_hook_path.name in content + assert prepend_hook_path.name in content assert 'hookB' not in content - # create_hook_append_value - hook_path = extension.create_hook_append_value( - 'append_env_hook_name', prefix_path, 'pkg_name', - 'APPEND_NAME', 'append_subdirectory') - assert hook_path.exists() - assert hook_path.name == 'append_env_hook_name.sh' - content = hook_path.read_text() - assert 'APPEND_NAME' in content - - # create_hook_prepend_value - hook_path = extension.create_hook_prepend_value( - 'env_hook_name', prefix_path, 'pkg_name', 'NAME', 'subdirectory') - assert hook_path.exists() - assert hook_path.name == 'env_hook_name.sh' - content = hook_path.read_text() - assert 'NAME' in content - # generate_command_environment if sys.platform == 'win32': with pytest.raises(SkipExtensionException) as e: @@ -94,3 +98,158 @@ def _test_extension(prefix_path): 'task_name', prefix_path, {'dep': str(prefix_path)}) env = run_until_complete(coroutine) assert isinstance(env, dict) + + subdirectory_path = str(prefix_path / 'subdirectory') + + # validate appending/prepending without existing values + with patch.dict(os.environ) as env_patch: + env_patch.pop('APPEND_NAME', None) + env_patch.pop('PREPEND_NAME', None) + coroutine = extension.generate_command_environment( + 'task_name', prefix_path, {'pkg_name': str(prefix_path)}) + env = run_until_complete(coroutine) + assert env.get('APPEND_NAME') == subdirectory_path + assert env.get('PREPEND_NAME') == subdirectory_path + + # validate appending/prepending with existing values + with patch.dict(os.environ, { + 'APPEND_NAME': 'control', + 'PREPEND_NAME': 'control', + }): + coroutine = extension.generate_command_environment( + 'task_name', prefix_path, {'pkg_name': str(prefix_path)}) + env = run_until_complete(coroutine) + # Expect excess separators to be removed + assert env.get('APPEND_NAME') == os.pathsep.join(( + 'control', + subdirectory_path, + )) + # Expect excess separators to be removed + assert env.get('PREPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + + # validate appending/prepending unique values + with patch.dict(os.environ, { + 'APPEND_NAME': os.pathsep.join((subdirectory_path, 'control')), + 'PREPEND_NAME': os.pathsep.join(('control', subdirectory_path)), + }): + coroutine = extension.generate_command_environment( + 'task_name', prefix_path, {'pkg_name': str(prefix_path)}) + env = run_until_complete(coroutine) + # Expect no change, value already appears earlier in the list + assert env.get('APPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + # Expect value to be *moved* to the front of the list + assert env.get('PREPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + + +async def _run_prefix_script(prefix_script): + # Attempt to use null-separated env output, but fall back to the + # best-effort implementation if not supported (i.e. older macOS) + cmd = ['.', str(prefix_script), '&&', 'env', '-0'] + try: + return await get_null_separated_environment_variables(cmd) + except AssertionError: + cmd.pop() + return await get_environment_variables(cmd) + + +def _test_prefix_script(prefix_path): + extension = ShShell() + + # ShShell requires the (non-primary) DsvShell extension as well + dsv_extension = DsvShell() + + # create_hook_append_value + append_hook_path = dsv_extension.create_hook_append_value( + 'append_env_hook_name', prefix_path, 'pkg_name', + 'APPEND_NAME', 'subdirectory') + assert append_hook_path.exists() + assert append_hook_path.name == 'append_env_hook_name.dsv' + content = append_hook_path.read_text() + assert 'APPEND_NAME' in content + + # create_hook_prepend_value + prepend_hook_path = dsv_extension.create_hook_prepend_value( + 'prepend_env_hook_name', prefix_path, 'pkg_name', + 'PREPEND_NAME', 'subdirectory') + assert prepend_hook_path.exists() + assert prepend_hook_path.name == 'prepend_env_hook_name.dsv' + content = prepend_hook_path.read_text() + assert 'PREPEND_NAME' in content + + # mark package as installed + package_index = prefix_path / get_relative_package_index_path() + package_index.mkdir(parents=True) + (package_index / 'pkg_name').write_text('') + + # create_package_script + dsv_extension.create_package_script( + prefix_path, 'pkg_name', [ + (append_hook_path.relative_to(prefix_path), ()), + (prepend_hook_path.relative_to(prefix_path), ())]) + assert (prefix_path / 'share' / 'pkg_name' / 'package.dsv').exists() + content = (prefix_path / 'share' / 'pkg_name' / 'package.dsv').read_text() + assert append_hook_path.name in content + assert prepend_hook_path.name in content + + # create_prefix_script + extension.create_prefix_script(prefix_path, True) + prefix_script = prefix_path / 'local_setup.sh' + prefix_script.exists() + assert (prefix_path / '_local_setup_util_sh.py').exists() + + if sys.platform != 'win32': + subdirectory_path = str(prefix_path / 'subdirectory') + + # validate appending/prepending without existing values + with patch.dict(os.environ, { + 'APPEND_NAME': '', + 'PREPEND_NAME': '', + }): + coroutine = _run_prefix_script(prefix_script) + env = run_until_complete(coroutine) + assert env.get('APPEND_NAME') == subdirectory_path + assert env.get('PREPEND_NAME') == subdirectory_path + + # validate appending/prepending with existing values + with patch.dict(os.environ, { + 'APPEND_NAME': 'control', + 'PREPEND_NAME': 'control', + }): + coroutine = _run_prefix_script(prefix_script) + env = run_until_complete(coroutine) + assert env.get('APPEND_NAME') == os.pathsep.join(( + 'control', + subdirectory_path, + )) + assert env.get('PREPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + + # validate appending/prepending unique values + with patch.dict(os.environ, { + 'APPEND_NAME': os.pathsep.join((subdirectory_path, 'control')), + 'PREPEND_NAME': os.pathsep.join(('control', subdirectory_path)), + }): + coroutine = _run_prefix_script(prefix_script) + env = run_until_complete(coroutine) + # Expect no change, value already appears earlier in the list + assert env.get('APPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + # TODO: The DsvShell behavior doesn't align with ShShell! + # ~~Expect value to be *moved* to the front of the list~~ + assert env.get('PREPEND_NAME') == os.pathsep.join(( + 'control', + subdirectory_path, + ))