Skip to content
Merged
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
5 changes: 5 additions & 0 deletions examples/posim_demos/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -8,4 +8,9 @@ install(
DESTINATION share/posim_demos
)

if(BUILD_TESTING)
find_package(ament_cmake_pytest REQUIRED)
ament_add_pytest_test(test_world_paths test/test_world_paths.py)
endif()

ament_package()
11 changes: 9 additions & 2 deletions examples/posim_demos/launch/posim_robot.launch.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import os
import shlex
import xml.etree.ElementTree as ET

from launch import LaunchDescription
Expand All @@ -11,6 +13,11 @@
from launch_ros.substitutions import FindPackageShare


def _quote_path(path):
# Keep Windows shell handling unchanged; shlex.quote is POSIX-only.
return shlex.quote(path) if os.name == "posix" else path


def launch_setup(context, *args, **kwargs):
paused = LaunchConfiguration("paused")
gui = LaunchConfiguration("gui")
Expand Down Expand Up @@ -47,10 +54,10 @@ def launch_setup(context, *args, **kwargs):
if world_element is None or not world_element.get("name"):
raise ValueError(f"World file [{world_filename}] does not declare a world name")
world_entity_name = world_element.get("name")
gz_args = [world_filepath]
gz_args = [_quote_path(world_filepath.perform(context))]
else:
world_entity_name = "empty"
gz_args = [world_name]
gz_args = [_quote_path(world_name.perform(context))]

zoom_camera_value = "true" if selected_world_name == "posim_ocean_waves" else "false"

Expand Down
8 changes: 7 additions & 1 deletion examples/posim_demos/launch/posim_world.launch.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,17 @@
import os
import shlex
from ament_index_python.packages import get_package_share_directory
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, OpaqueFunction, IncludeLaunchDescription
from launch.launch_description_sources import PythonLaunchDescriptionSource
from launch.substitutions import LaunchConfiguration


def _quote_path(path):
# Keep Windows shell handling unchanged; shlex.quote is POSIX-only.
return shlex.quote(path) if os.name == "posix" else path


def launch_setup(context, *args, **kwargs):

pkg_posim_worlds = get_package_share_directory("posim_worlds")
Expand All @@ -19,7 +25,7 @@ def launch_setup(context, *args, **kwargs):
world_path = os.path.join(pkg_posim_worlds, "worlds", world_file_name)

# Gazebo simulation launch
gz_args = f"-r {world_path}"
gz_args = f"-r {_quote_path(world_path)}"
if verbose_flag.lower() == "true":
gz_args += " --verbose"
if headless_flag.lower() == "true":
Expand Down
4 changes: 4 additions & 0 deletions examples/posim_demos/package.xml
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,10 @@
<maintainer email="kys50728@g.kmou.ac.kr">Ye-Seol Kwon</maintainer>
<license>Apache 2.0</license>
<buildtool_depend>ament_cmake</buildtool_depend>
<test_depend>ament_cmake_pytest</test_depend>
<test_depend>ament_index_python</test_depend>
<test_depend>launch</test_depend>
<test_depend>launch_ros</test_depend>
<export>
<build_type>ament_cmake</build_type>
</export>
Expand Down
116 changes: 116 additions & 0 deletions examples/posim_demos/test/test_world_paths.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
"""Exercise world-path quoting at both launch entry points."""

from contextlib import ExitStack
from importlib.util import module_from_spec, spec_from_file_location
import json
import os
from pathlib import Path
import shlex
import subprocess
import sys
from types import SimpleNamespace
from unittest.mock import patch

from launch import LaunchContext
from launch.actions import DeclareLaunchArgument
from launch.utilities import normalize_to_list_of_substitutions, perform_substitutions
import pytest


def arguments(entry, share, platform_name="posix", empty=False, headless=True, paused=False):
path = Path(__file__).resolve().parents[1] / "launch" / f"posim_{entry}.launch.py"
spec = spec_from_file_location("launch_under_test", path)
module = module_from_spec(spec)
spec.loader.exec_module(module)
module.os = SimpleNamespace(**{**os.__dict__, "name": platform_name})
context = LaunchContext()
for action in module.generate_launch_description().entities:
if isinstance(action, DeclareLaunchArgument) and action.default_value is not None:
context.launch_configurations[action.name] = perform_substitutions(
context, action.default_value
)
context.launch_configurations.update(
world_name="empty.sdf" if empty else "scene",
headless=str(headless).lower(),
gui=str(not headless).lower(),
paused=str(paused).lower(),
verbose="4" if entry == "robot" else "true",
)
with ExitStack() as stack:
if entry == "robot":
stack.enter_context(
patch.object(module.FindPackageShare, "find", lambda self, package: str(share))
)
else:
stack.enter_context(
patch.object(module, "get_package_share_directory", return_value=str(share))
)
actions = module.launch_setup(context)
include = next(a for a in actions if "gz_args" in dict(a.launch_arguments))
value = dict(include.launch_arguments)["gz_args"]
return perform_substitutions(context, normalize_to_list_of_substitutions(value))


@pytest.fixture
def share(tmp_path):
path = tmp_path / "plain"
(path / "worlds").mkdir(parents=True)
(path / "worlds/scene.world").write_text('<sdf><world name="scene"/></sdf>')
return path


@pytest.mark.skipif(os.name != "posix", reason="Requires a POSIX shell")
@pytest.mark.parametrize("entry", ["world", "robot"])
@pytest.mark.parametrize(
"directory",
[
"plain",
"space path",
"author's files",
"semi;colon",
"literal$HOME",
'double"quote',
"해양 연구",
],
)
def test_world_remains_one_literal_shell_argument(tmp_path, entry, directory):
share = tmp_path / directory
(share / "worlds").mkdir(parents=True)
world = share / "worlds/scene.world"
world.write_text('<sdf><world name="scene"/></sdf>')
prefix = (
shlex.quote(sys.executable)
+ " -c "
+ shlex.quote("import json,sys; print(json.dumps(sys.argv[1:]))")
)
result = subprocess.run(
prefix + " " + arguments(entry, share),
shell=True,
executable="/bin/sh",
capture_output=True,
text=True,
timeout=10,
)
assert result.returncode == 0, result.stderr
argv = json.loads(result.stdout)
assert argv.count(str(world)) == 1
assert "-r" in argv
assert "-s" in argv


@pytest.mark.parametrize("entry", ["world", "robot"])
def test_windows_keeps_unquoted_command_construction(share, entry):
# Deliberately preserve Windows behavior; not a Windows execution test.
spaced = share.with_name("Program Files")
share.rename(spaced)
assert str(spaced / "worlds/scene.world") in arguments(entry, spaced, "nt")
assert "'" not in arguments(entry, spaced, "nt")


@pytest.mark.parametrize("headless", [False, True])
@pytest.mark.parametrize("paused", [False, True])
def test_robot_empty_world_preserves_flags(share, headless, paused):
argv = shlex.split(arguments("robot", share, empty=True, headless=headless, paused=paused))
assert argv.count("empty.sdf") == 1
assert ("-r" in argv) == (not paused)
assert ("-s" in argv) == headless
Loading