diff --git a/colcon_core/shell/dsv.py b/colcon_core/shell/dsv.py index 2118d545..b2797942 100644 --- a/colcon_core/shell/dsv.py +++ b/colcon_core/shell/dsv.py @@ -1,19 +1,35 @@ # Copyright 2019 Dirk Thomas # Licensed under the Apache License, Version 2.0 +import os from pathlib import Path +from colcon_core.environment_variable import EnvironDict from colcon_core.plugin_system import satisfies_version +from colcon_core.plugin_system import SkipExtensionException +from colcon_core.shell import _get_pkg_name +from colcon_core.shell import check_dependency_availability +from colcon_core.shell import get_shell_extensions from colcon_core.shell import logger from colcon_core.shell import ShellExtensionPoint from colcon_core.shell.template import expand_template +DSV_TYPE_APPEND_NON_DUPLICATE = 'append-non-duplicate' +DSV_TYPE_PREPEND_NON_DUPLICATE = 'prepend-non-duplicate' +DSV_TYPE_PREPEND_NON_DUPLICATE_IF_EXISTS = 'prepend-non-duplicate-if-exists' +DSV_TYPE_SET = 'set' +DSV_TYPE_SET_IF_UNSET = 'set-if-unset' +DSV_TYPE_SOURCE = 'source' + class DsvShell(ShellExtensionPoint): - """Generate `.dsv` files describing the intended environment change.""" + """Generate `.dsv` files and derive the command environment from them.""" # the priority needs to be higher than the default for primary shells - PRIORITY = 200 + # it is also higher than the priority of any shell which computes the + # command environment by invoking a subprocess, since the descriptors + # provide the same information without spawning one + PRIORITY = 400 def __init__(self): # noqa: D107 super().__init__() @@ -81,3 +97,199 @@ def create_hook_prepend_value( # noqa: D102 'value': subdirectory, }) return hook_path + + async def generate_command_environment( # noqa: D102 + self, task_name, build_base, dependencies, + ): + try: + # check if all dependencies are available + # removes dependencies available in the environment from the + # parameter + check_dependency_availability( + dependencies, script_filename='package.dsv') + except RuntimeError as e: # noqa: F841 + # a prefix without descriptors can still be sourced by a shell, + # which also reports the packages which need to be built + raise SkipExtensionException( + 'Not all dependencies provide a descriptor') + + env = EnvironDict(os.environ) + try: + for dep, pkg_install_base in dependencies.items(): + prefix = str(pkg_install_base) + _evaluate_dsv_file( + env, + os.path.join( + prefix, 'share', _get_pkg_name(dep), 'package.dsv'), + prefix) + except _MissingDescriptorError as e: # noqa: F841 + raise SkipExtensionException( + f'Unable to determine the environment from descriptors: {e}') + + # the package scripts unset this variable after being sourced + env.pop('COLCON_CURRENT_PREFIX', None) + + # write environment variables to file for debugging + # the other shells create the directory with their command prefix + # script, which this extension doesn't generate + env_path = build_base / ( + 'colcon_command_prefix_%s.dsv.env' % task_name) + env_path.parent.mkdir(parents=True, exist_ok=True) + with env_path.open('w') as h: + for key in sorted(env.keys()): + value = env[key] + h.write(f'{key}={value}\n') + + return env + + +class _MissingDescriptorError(Exception): + """A script which needs to be sourced has no descriptor to evaluate.""" + + +_script_extensions = None + + +def _get_script_extensions(): + """Get the file extensions which a primary shell is able to source.""" + global _script_extensions + if _script_extensions is None: + _script_extensions = set() + extensions = get_shell_extensions() + for priority in extensions.keys(): + # only consider primary shell extensions + if priority <= ShellExtensionPoint.PRIORITY: + break + for extension in extensions[priority].values(): + _script_extensions.update(extension.get_file_extensions()) + return _script_extensions + + +def _parse_dsv_file(dsv_path): + """Parse a descriptor into a list of type and remainder tuples.""" + entries = [] + try: + with open(dsv_path, 'r') as h: + lines = h.read().splitlines() + except OSError as e: # noqa: F841 + raise _MissingDescriptorError(f"could not read '{dsv_path}': {e}") + for i, line in enumerate(lines): + # skip over empty or whitespace-only lines as well as comments + if not line.strip() or line.startswith('#'): + continue + try: + type_, remainder = line.split(';', 1) + except ValueError: + raise RuntimeError( + "Line %d in '%s' doesn't contain a semicolon separating the " + 'type from the arguments' % (i + 1, dsv_path)) + entries.append((type_, remainder)) + + return entries + + +def _flatten_dsv_file(dsv_path, prefix): + """ + Resolve a descriptor and the descriptors it sources into a list of changes. + + This mirrors the `process_dsv_file` function of the `prefix_util.py` + template, but instead of generating shell commands it collects the + environment changes in the order in which they need to be applied. + + :param str dsv_path: The path of the descriptor + :param str prefix: The path of the install prefix + :returns: The type and remainder tuples to be applied in order + :rtype: list + :raises _MissingDescriptorError: if a script needs to be sourced for which + no descriptor exists + """ + changes = [] + basename_map = {} + for type_, remainder in _parse_dsv_file(dsv_path): + if type_ != DSV_TYPE_SOURCE: + # handle non-source lines + changes.append((type_, remainder)) + continue + # group remaining source lines by basename + path_without_ext, ext = os.path.splitext(remainder) + extensions = basename_map.setdefault(path_without_ext, set()) + # the descriptor of a basename is considered either way, only script + # extensions which a shell would source are relevant here + if ext[1:] in _get_script_extensions(): + extensions.add(ext[1:]) + + for basename, extensions in basename_map.items(): + if not os.path.isabs(basename): + basename = os.path.join(prefix, basename) + if os.path.exists(basename + '.dsv'): + # process descriptors recursively + changes += _flatten_dsv_file(basename + '.dsv', prefix) + elif extensions: + # a script without a descriptor can only be sourced by a shell + raise _MissingDescriptorError( + "no descriptor for '%s.%s'" + % (basename, sorted(extensions)[0])) + + return changes + + +def _evaluate_dsv_file(env, dsv_path, prefix): + """Apply a descriptor and the descriptors it sources to an environment.""" + for type_, remainder in _flatten_dsv_file(dsv_path, prefix): + try: + _apply_change(env, type_, remainder, prefix) + except RuntimeError as e: + raise RuntimeError("In '%s' %s" % (dsv_path, e)) from e + + +def _join_with_prefix(prefix, value): + # the hook scripts are generated with native separators, so a descriptor + # using the alternative separator must not result in a different value + if os.altsep: + value = value.replace(os.altsep, os.sep) + return os.path.join(prefix, value) + + +def _apply_change(env, type_, remainder, prefix): + if type_ in (DSV_TYPE_SET, DSV_TYPE_SET_IF_UNSET): + try: + env_name, value = remainder.split(';', 1) + except ValueError: + raise RuntimeError( + "doesn't contain a semicolon separating the environment name " + 'from the value') + try_prefixed_value = \ + _join_with_prefix(prefix, value) if value else prefix + if os.path.exists(try_prefixed_value): + value = try_prefixed_value + if type_ == DSV_TYPE_SET or not env.get(env_name): + env[env_name] = value + elif type_ in ( + DSV_TYPE_APPEND_NON_DUPLICATE, + DSV_TYPE_PREPEND_NON_DUPLICATE, + DSV_TYPE_PREPEND_NON_DUPLICATE_IF_EXISTS + ): + env_name_and_values = remainder.split(';') + env_name = env_name_and_values[0] + for value in env_name_and_values[1:]: + if not value: + value = prefix + elif not os.path.isabs(value): + value = _join_with_prefix(prefix, value) + if ( + type_ == DSV_TYPE_PREPEND_NON_DUPLICATE_IF_EXISTS and + not os.path.exists(value) + ): + continue + # duplicates are being skipped and empty values dropped, the same + # way the `prefix_util.py` template evaluates the descriptors + values = [v for v in env.get(env_name, '').split(os.pathsep) if v] + if value not in values: + if type_ == DSV_TYPE_APPEND_NON_DUPLICATE: + values.append(value) + else: + values.insert(0, value) + env[env_name] = os.pathsep.join(values) + else: + raise RuntimeError( + 'contains an unknown environment hook type: ' + type_) diff --git a/test/test_shell_dsv.py b/test/test_shell_dsv.py new file mode 100644 index 00000000..cd7a4afd --- /dev/null +++ b/test/test_shell_dsv.py @@ -0,0 +1,215 @@ +# Copyright 2026 Open Source Robotics Foundation, Inc. +# Licensed under the Apache License, Version 2.0 + +import os +from unittest.mock import patch + +from colcon_core.plugin_system import SkipExtensionException +from colcon_core.shell.dsv import DsvShell +import pytest + +from .run_until_complete import run_until_complete + + +def _write_package_descriptor(prefix_path, pkg_name, lines): + descriptor = prefix_path / 'share' / pkg_name / 'package.dsv' + descriptor.parent.mkdir(parents=True, exist_ok=True) + descriptor.write_text(''.join(f'{line}\n' for line in lines)) + return descriptor + + +def _generate_command_environment(prefix_path, pkg_name): + extension = DsvShell() + coroutine = extension.generate_command_environment( + 'task_name', prefix_path, {pkg_name: str(prefix_path)}) + return run_until_complete(coroutine) + + +def test_extension(tmp_path): + extension = DsvShell() + + # create_prefix_script + assert extension.create_prefix_script(tmp_path, True) == [] + + # create_hook_append_value + append_hook_path = extension.create_hook_append_value( + 'append_env_hook_name', tmp_path, 'pkg_name', + 'APPEND_NAME', 'subdirectory') + assert append_hook_path.exists() + assert append_hook_path.name == 'append_env_hook_name.dsv' + assert 'APPEND_NAME' in append_hook_path.read_text() + + # create_hook_prepend_value + prepend_hook_path = extension.create_hook_prepend_value( + 'prepend_env_hook_name', tmp_path, 'pkg_name', + 'PREPEND_NAME', 'subdirectory') + assert prepend_hook_path.exists() + assert prepend_hook_path.name == 'prepend_env_hook_name.dsv' + assert 'PREPEND_NAME' in prepend_hook_path.read_text() + + # create_hook_set_value + set_hook_path = extension.create_hook_set_value( + 'set_env_hook_name', tmp_path, 'pkg_name', 'SET_NAME', 'value') + assert set_hook_path.exists() + assert set_hook_path.name == 'set_env_hook_name.dsv' + assert 'SET_NAME' in set_hook_path.read_text() + + # create_package_script + extension.create_package_script( + tmp_path, 'pkg_name', [ + (append_hook_path.relative_to(tmp_path), ()), + (prepend_hook_path.relative_to(tmp_path), ()), + (set_hook_path.relative_to(tmp_path), ())]) + descriptor = tmp_path / 'share' / 'pkg_name' / 'package.dsv' + assert descriptor.exists() + content = descriptor.read_text() + assert append_hook_path.name in content + assert prepend_hook_path.name in content + assert set_hook_path.name in content + + +def test_generate_command_environment(tmp_path): + extension = DsvShell() + + # dependency descriptor missing, a shell extension needs to take over + with pytest.raises(SkipExtensionException) as e: + coroutine = extension.generate_command_environment( + 'task_name', tmp_path, {'dep': str(tmp_path)}) + run_until_complete(coroutine) + assert str(e.value) == 'Not all dependencies provide a descriptor' + + # dependency descriptor exists + _write_package_descriptor(tmp_path, 'dep', []) + with patch.dict(os.environ, {'CONTROL_NAME': 'control'}): + env = _generate_command_environment(tmp_path, 'dep') + assert isinstance(env, dict) + assert env.get('CONTROL_NAME') == 'control' + + # the environment is written for debugging + env_path = tmp_path / 'colcon_command_prefix_task_name.dsv.env' + assert 'CONTROL_NAME=control' in env_path.read_text().splitlines() + + # the build base doesn't exist yet in a clean build + build_base = tmp_path / 'build' / 'dep' + coroutine = extension.generate_command_environment( + 'task_name', build_base, {'dep': str(tmp_path)}) + run_until_complete(coroutine) + assert (build_base / 'colcon_command_prefix_task_name.dsv.env').exists() + + +def test_generate_command_environment_values(tmp_path): + subdirectory_path = str(tmp_path / 'subdirectory') + _write_package_descriptor(tmp_path, 'pkg_name', [ + 'append-non-duplicate;APPEND_NAME;subdirectory', + 'prepend-non-duplicate;PREPEND_NAME;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) + env = _generate_command_environment(tmp_path, 'pkg_name') + 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', + }): + env = _generate_command_environment(tmp_path, 'pkg_name') + 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)), + }): + env = _generate_command_environment(tmp_path, 'pkg_name') + # Expect no change, value already appears in the list + assert env.get('APPEND_NAME') == os.pathsep.join(( + subdirectory_path, + 'control', + )) + # Expect no change either, the descriptors are being evaluated like the + # `prefix_util.py` template does, which skips duplicates instead of moving + # them to the front of the list like the shell specific hook scripts do + assert env.get('PREPEND_NAME') == os.pathsep.join(( + 'control', + subdirectory_path, + )) + + +def test_generate_command_environment_types(tmp_path): + (tmp_path / 'subdirectory').mkdir() + _write_package_descriptor(tmp_path, 'pkg_name', [ + '# a comment', + '', + 'set;SET_NAME;subdirectory', + 'set;SET_OTHER_NAME;not-a-path', + 'set-if-unset;SET_IF_UNSET_NAME;value', + 'set-if-unset;CONTROL_NAME;value', + 'prepend-non-duplicate-if-exists;IF_EXISTS_NAME;subdirectory', + 'prepend-non-duplicate-if-exists;IF_EXISTS_NAME;other-subdirectory', + ]) + + with patch.dict(os.environ, {'CONTROL_NAME': 'control'}): + env = _generate_command_environment(tmp_path, 'pkg_name') + + # a value which is a path relative to the prefix is being expanded + assert env.get('SET_NAME') == str(tmp_path / 'subdirectory') + assert env.get('SET_OTHER_NAME') == 'not-a-path' + assert env.get('SET_IF_UNSET_NAME') == 'value' + assert env.get('CONTROL_NAME') == 'control' + # only the existing path is being prepended + assert env.get('IF_EXISTS_NAME') == str(tmp_path / 'subdirectory') + + # an unknown type is an error + _write_package_descriptor( + tmp_path, 'pkg_name', ['unknown-type;NAME;value']) + with pytest.raises(RuntimeError) as e: + _generate_command_environment(tmp_path, 'pkg_name') + assert 'unknown environment hook type' in str(e.value) + + # a line without a semicolon is an error + _write_package_descriptor(tmp_path, 'pkg_name', ['no-semicolon']) + with pytest.raises(RuntimeError) as e: + _generate_command_environment(tmp_path, 'pkg_name') + assert 'semicolon separating the type' in str(e.value) + + +def test_generate_command_environment_recursion(tmp_path): + hook_path = tmp_path / 'share' / 'pkg_name' / 'hook' / 'hook_name.dsv' + hook_path.parent.mkdir(parents=True) + hook_path.write_text('set;SET_NAME;value\n') + _write_package_descriptor(tmp_path, 'pkg_name', [ + # scripts of other shells are only considered by their descriptor + 'source;share/pkg_name/hook/hook_name.ext', + 'source;share/pkg_name/hook/hook_name.dsv', + # a basename without any descriptor or known script is being ignored + 'source;share/pkg_name/hook/other_hook_name.marker', + ]) + + with patch( + 'colcon_core.shell.dsv._get_script_extensions', return_value={'ext'} + ): + env = _generate_command_environment(tmp_path, 'pkg_name') + assert env.get('SET_NAME') == 'value' + + # a script of another shell without a descriptor can't be evaluated + _write_package_descriptor(tmp_path, 'pkg_name', [ + 'source;share/pkg_name/hook/script_only.ext', + ]) + with patch( + 'colcon_core.shell.dsv._get_script_extensions', return_value={'ext'} + ): + with pytest.raises(SkipExtensionException) as e: + _generate_command_environment(tmp_path, 'pkg_name') + assert 'script_only.ext' in str(e.value)