Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 37 additions & 2 deletions launch/launch/actions/include_launch_description.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@
import launch.logging

from .opaque_function import OpaqueFunction
from .pop_environment import PopEnvironment
from .pop_launch_configurations import PopLaunchConfigurations
from .push_environment import PushEnvironment
from .push_launch_configurations import PushLaunchConfigurations
from .set_launch_configuration import SetLaunchConfiguration
from ..action import Action
from ..frontend import Entity
Expand Down Expand Up @@ -126,17 +130,31 @@ def __init__(
self,
launch_description_source: Union[LaunchDescriptionSource, SomeSubstitutionsType],
*,
scoped: bool = False,
launch_arguments: Optional[
Iterable[Tuple[SomeSubstitutionsType, SomeSubstitutionsType]]
] = None,
**kwargs: Any
) -> None:
"""Create an IncludeLaunchDescription action."""
"""
Create an IncludeLaunchDescription action.

When `scoped` is True, launch configurations and environment variables set by the
included launch description are isolated to the scope of the include and will not
propagate back to the parent context. The included description still has access to
the parent's existing launch configurations (forwarding is always enabled for scoped
includes).

:param scoped: if True, push/pop launch configurations and environment around the
included description so that `SetLaunchConfiguration` and environment changes
do not leak into the parent scope. Defaults to False for backward compatibility.
"""
super().__init__(**kwargs)
if not isinstance(launch_description_source, LaunchDescriptionSource):
launch_description_source = AnyLaunchDescriptionSource(launch_description_source)
self.__launch_description_source = launch_description_source
self.__launch_arguments = () if launch_arguments is None else tuple(launch_arguments)
self.__scoped = scoped
self.__logger = launch.logging.get_logger(__name__)

@classmethod
Expand All @@ -146,6 +164,9 @@ def parse(cls, entity: Entity, parser: Parser
_, kwargs = super().parse(entity, parser)
file_path = parser.parse_substitution(entity.get_attr('file'))
kwargs['launch_description_source'] = file_path
scoped = entity.get_attr('scoped', data_type=bool, optional=True)
if scoped is not None:
kwargs['scoped'] = scoped
args = []
args_arg = entity.get_attr('arg', data_type=List[Entity], optional=True)
if args_arg is not None:
Expand Down Expand Up @@ -249,12 +270,26 @@ def execute(self, context: LaunchContext) -> List[Union[SetLaunchConfiguration,
set_launch_configuration_actions.append(SetLaunchConfiguration(name, value))

# Set launch arguments as launch configurations and then include the launch description.
return [
actions = [
*set_launch_configuration_actions,
launch_description,
OpaqueFunction(function=self._restore_launch_file_location_locals),
]

if self.__scoped:
# Wrap with push/pop to isolate launch configurations and environment changes.
# Forwarding is always enabled: the included description sees the parent's
# existing launch configurations but its mutations do not leak back.
return [
PushLaunchConfigurations(),
PushEnvironment(),
*actions,
PopEnvironment(),
PopLaunchConfigurations(),
]

return actions

def _set_launch_file_location_locals(self, context: LaunchContext) -> None:
context._push_locals()
# Keep the previous launch file path/dir locals so that we can restore them after
Expand Down
176 changes: 176 additions & 0 deletions launch/test/launch/actions/test_include_launch_description.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,10 @@
from launch.actions import DeclareLaunchArgument
from launch.actions import IncludeLaunchDescription
from launch.actions import OpaqueFunction
from launch.actions import PopEnvironment
from launch.actions import PopLaunchConfigurations
from launch.actions import PushEnvironment
from launch.actions import PushLaunchConfigurations
from launch.actions import ResetLaunchConfigurations
from launch.actions import SetEnvironmentVariable
from launch.actions import SetLaunchConfiguration
Expand All @@ -34,6 +38,8 @@

import pytest

from temporary_environment import sandbox_environment_variables


def test_include_launch_description_constructors():
"""Test the constructors for IncludeLaunchDescription class."""
Expand Down Expand Up @@ -259,6 +265,176 @@ def test_include_launch_description_launch_arguments():
action2.visit(lc2)


@sandbox_environment_variables
def test_include_launch_description_scoped_execute():
"""Test scoped=True: Push/Pop wrapping, forwarding, and isolation of launch configurations."""
ld_child = LaunchDescription([])
action = IncludeLaunchDescription(
LaunchDescriptionSource(ld_child),
launch_arguments={'bar': 'BAR'}.items(),
scoped=True,
)

lc = LaunchContext()
lc.launch_configurations['foo'] = 'FOO'

result = action.visit(lc)

# Expected: Push, Push, SetLaunchConfig, LaunchDescription, OpaqueFunction, Pop, Pop
assert len(result) == 7
assert isinstance(result[0], PushLaunchConfigurations)
assert isinstance(result[1], PushEnvironment)
assert isinstance(result[2], SetLaunchConfiguration)
assert result[3] == ld_child
assert isinstance(result[4], OpaqueFunction)
assert isinstance(result[5], PopEnvironment)
assert isinstance(result[6], PopLaunchConfigurations)

# Step through and verify intermediate state
result[0].visit(lc) # PushLaunchConfigurations
assert lc.launch_configurations['foo'] == 'FOO' # forwarded to child scope

result[1].visit(lc) # PushEnvironment

result[2].visit(lc) # SetLaunchConfiguration('bar', 'BAR')
assert lc.launch_configurations['bar'] == 'BAR'
assert lc.launch_configurations['foo'] == 'FOO' # still visible

# Simulate what the child launch description would do
lc.launch_configurations['baz'] = 'BAZ'
assert lc.launch_configurations['baz'] == 'BAZ'

# result[3] (LaunchDescription) and result[4] (OpaqueFunction) skipped — they don't affect
# launch_configurations directly in this test

result[5].visit(lc) # PopEnvironment
result[6].visit(lc) # PopLaunchConfigurations
# After pop, child's configs are gone, parent's are restored
assert lc.launch_configurations['foo'] == 'FOO'
assert 'baz' not in lc.launch_configurations
assert 'bar' not in lc.launch_configurations
assert len(lc.launch_configurations) == 1


@sandbox_environment_variables
def test_include_launch_description_unscoped_execute():
"""Test scoped=False (default): no Push/Pop, configurations leak to parent."""
ld_child = LaunchDescription([])
action = IncludeLaunchDescription(
LaunchDescriptionSource(ld_child),
launch_arguments={'bar': 'BAR'}.items(),
)

lc = LaunchContext()
lc.launch_configurations['foo'] = 'FOO'

result = action.visit(lc)

# Expected: SetLaunchConfig, LaunchDescription, OpaqueFunction (no Push/Pop)
assert len(result) == 3
assert isinstance(result[0], SetLaunchConfiguration)
assert result[1] == ld_child
assert isinstance(result[2], OpaqueFunction)
assert not any(isinstance(r, PushLaunchConfigurations) for r in result)
assert not any(isinstance(r, PopLaunchConfigurations) for r in result)

# Step through
result[0].visit(lc) # SetLaunchConfiguration('bar', 'BAR')
assert lc.launch_configurations['bar'] == 'BAR'
assert lc.launch_configurations['foo'] == 'FOO' # untouched

# After all actions, bar persists — it leaked to the parent scope
assert len(lc.launch_configurations) == 2
assert lc.launch_configurations['bar'] == 'BAR'


@sandbox_environment_variables
def test_include_launch_description_scoped_isolates_environment():
"""Test scoped=True: environment variable changes do not leak to parent."""
ld_child = LaunchDescription([])
action = IncludeLaunchDescription(
LaunchDescriptionSource(ld_child),
scoped=True,
)

lc = LaunchContext()
assert 'env_foo' not in lc.environment

result = action.visit(lc)

assert isinstance(result[0], PushLaunchConfigurations)
assert isinstance(result[1], PushEnvironment)

result[0].visit(lc) # PushLaunchConfigurations
result[1].visit(lc) # PushEnvironment

# Simulate child setting an environment variable
lc.environment['env_foo'] = 'FOO'
assert lc.environment['env_foo'] == 'FOO'

assert isinstance(result[-2], PopEnvironment)
assert isinstance(result[-1], PopLaunchConfigurations)

result[-2].visit(lc) # PopEnvironment
assert 'env_foo' not in lc.environment # rolled back

result[-1].visit(lc) # PopLaunchConfigurations


@sandbox_environment_variables
def test_include_launch_description_unscoped_leaks_environment():
"""Test scoped=False (default): environment variable changes leak to parent."""
ld_child = LaunchDescription([])
action = IncludeLaunchDescription(
LaunchDescriptionSource(ld_child),
)

lc = LaunchContext()
result = action.visit(lc)

# No Push/Pop — environment mutations persist
assert len(result) == 2 # LaunchDescription, OpaqueFunction (no launch_arguments)

# Simulate child setting an environment variable
lc.environment['env_foo'] = 'FOO'

# After all actions, the env var persists — no Pop to roll it back
assert lc.environment['env_foo'] == 'FOO'


@sandbox_environment_variables
def test_include_launch_description_scoped_with_overwrite():
"""Test scoped=True: child overwrites parent config, but parent value is restored after pop."""
ld_child = LaunchDescription([])
action = IncludeLaunchDescription(
LaunchDescriptionSource(ld_child),
launch_arguments={'foo': 'OOF'}.items(),
scoped=True,
)

lc = LaunchContext()
lc.launch_configurations['foo'] = 'FOO'
lc.launch_configurations['bar'] = 'BAR'

result = action.visit(lc)

result[0].visit(lc) # PushLaunchConfigurations
assert lc.launch_configurations['foo'] == 'FOO' # copied to new scope
assert lc.launch_configurations['bar'] == 'BAR' # forwarded

result[1].visit(lc) # PushEnvironment

result[2].visit(lc) # SetLaunchConfiguration('foo', 'OOF')
assert lc.launch_configurations['foo'] == 'OOF' # overwritten in child scope
assert lc.launch_configurations['bar'] == 'BAR' # untouched

result[-2].visit(lc) # PopEnvironment
result[-1].visit(lc) # PopLaunchConfigurations
assert lc.launch_configurations['foo'] == 'FOO' # restored
assert lc.launch_configurations['bar'] == 'BAR' # still there
assert len(lc.launch_configurations) == 2


def test_include_python():
"""Test including Python, with and without explicit PythonLaunchDescriptionSource."""
this_dir = Path(__file__).parent
Expand Down
73 changes: 73 additions & 0 deletions launch_xml/test/launch_xml/test_include.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,5 +54,78 @@ def test_include():
assert ls.context.launch_configurations['baz'] == 'BAZ'


def test_include_scoped_true():
"""Parse include with scoped="true" — child configs do not leak to parent."""
path = (Path(__file__).parent / 'executable.xml').as_posix()
xml_file = \
"""\
<launch>
<let name="bar" value="BAR" />
<include file="{}" scoped="true">
<let name="foo" value="FOO" />
</include>
</launch>
""".format(path) # noqa: E501
xml_file = textwrap.dedent(xml_file)
root_entity, parser = load_no_extensions(io.StringIO(xml_file))
ld = parser.parse_description(root_entity)
include = ld.entities[1]
assert isinstance(include, IncludeLaunchDescription)
ls = LaunchService(debug=True)
ls.include_launch_description(ld)
assert 0 == ls.run()
# bar persists, but foo from scoped include does not leak
assert ls.context.launch_configurations['bar'] == 'BAR'
assert 'foo' not in ls.context.launch_configurations


def test_include_scoped_false():
"""Parse include with scoped="false" — child configs leak to parent (default behavior)."""
path = (Path(__file__).parent / 'executable.xml').as_posix()
xml_file = \
"""\
<launch>
<let name="bar" value="BAR" />
<include file="{}" scoped="false">
<let name="foo" value="FOO" />
</include>
</launch>
""".format(path) # noqa: E501
xml_file = textwrap.dedent(xml_file)
root_entity, parser = load_no_extensions(io.StringIO(xml_file))
ld = parser.parse_description(root_entity)
include = ld.entities[1]
assert isinstance(include, IncludeLaunchDescription)
ls = LaunchService(debug=True)
ls.include_launch_description(ld)
assert 0 == ls.run()
# Both bar and foo are visible
assert ls.context.launch_configurations['bar'] == 'BAR'
assert ls.context.launch_configurations['foo'] == 'FOO'


def test_include_default_is_unscoped():
"""Parse include without scoped attribute — defaults to unscoped (backward compatible)."""
path = (Path(__file__).parent / 'executable.xml').as_posix()
xml_file = \
"""\
<launch>
<include file="{}">
<let name="foo" value="FOO" />
</include>
</launch>
""".format(path) # noqa: E501
xml_file = textwrap.dedent(xml_file)
root_entity, parser = load_no_extensions(io.StringIO(xml_file))
ld = parser.parse_description(root_entity)
include = ld.entities[0]
assert isinstance(include, IncludeLaunchDescription)
ls = LaunchService(debug=True)
ls.include_launch_description(ld)
assert 0 == ls.run()
# foo leaks, same as before
assert ls.context.launch_configurations['foo'] == 'FOO'


if __name__ == '__main__':
test_include()
Loading