import importlib.util import signal import subprocess import unittest from pathlib import Path from unittest import mock MODULE_PATH = Path(__file__).parents[1] / "tools" / "launcher_ui.py" SPEC = importlib.util.spec_from_file_location("launcher_ui", MODULE_PATH) launcher_ui = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None SPEC.loader.exec_module(launcher_ui) class LauncherCommandsTest(unittest.TestCase): def test_modes_expose_the_confirmed_command_matrix(self) -> None: expected_titles = { "Simulation": [ "Dual Arm Mock Launch", "XRobotoolkit UDP Bridge (90 Hz)", "Sample UDP Sender (Both Staggered, 60s)", "Open Controller Hz Monitor", "Open ROS Topic/Node List Monitor", "Open Controller Topic Monitor", ], "MuJoCo": [ "Dual Arm MuJoCo Mock Launch", "Dual Arm MuJoCo Real Hardware Launch", "XRobotoolkit UDP Bridge (90 Hz)", "Open Controller Hz Monitor", "Open ROS Topic/Node List Monitor", "Open Controller Topic Monitor", ], "Real Hardware": [ "Ping Left RM75", "Ping Right RM75", "Left Arm RealMan Launch", "Right Arm RealMan Launch", "Dual Arm RealMan Launch", "XRobotoolkit UDP Bridge (90 Hz)", "Left Gripper Open", "Left Gripper Close", "Right Gripper Open", "Right Gripper Close", "Open ROS Topic/Node List Monitor", "Open Controller Topic Monitor", ], "Diagnostics": [ "ROS Doctor Report", "XR-RM Bringup Prefix", "XR-RM Input Prefix", "XR-RM Teleop Prefix", "XR-RM MuJoCo Prefix", "Open Controller Position Monitor", "Open Controller Hz Monitor", "Open ROS Topic/Node List Monitor", "Open Controller Topic Monitor", ], } self.assertEqual(launcher_ui.MODES, list(expected_titles)) for mode, titles in expected_titles.items(): actual = [ indexed_title.split(". ", 1)[1] for indexed_title, _command in launcher_ui.build_commands_by_mode(mode) ] self.assertEqual(actual, titles) def test_mujoco_launches_distinguish_mock_and_real_hardware(self) -> None: commands = dict(launcher_ui.build_commands_by_mode("MuJoCo")) self.assertIn( "arm:=both use_mock:=true use_mujoco:=true", commands["1. Dual Arm MuJoCo Mock Launch"], ) self.assertIn( "arm:=both use_mock:=false use_mujoco:=true", commands["2. Dual Arm MuJoCo Real Hardware Launch"], ) def test_cmd_vel_monitor_is_completely_removed(self) -> None: self.assertFalse(hasattr(launcher_ui, "CMD_VEL_MONITOR_ACTION")) for mode in launcher_ui.MODES: self.assertNotIn("cmd_vel", repr(launcher_ui.build_commands_by_mode(mode))) def test_environment_check_includes_mujoco_package(self) -> None: app = object.__new__(launcher_ui.LauncherApp) app.workspace_root = launcher_ui._find_workspace_root() app.terminal_command = lambda _title, _script: ["terminal"] app.x_terminal_target = lambda: "terminator" checked_packages = [] app._ros_package_available = lambda package: checked_packages.append(package) or True app.show_text_dialog = lambda *_args: None app.status = mock.Mock() with mock.patch.object( launcher_ui.shutil, "which", return_value="/usr/bin/x-terminal-emulator", ): app.check_prerequisites() self.assertEqual( checked_packages, ["xr_rm_bringup", "xr_rm_input", "xr_rm_teleop", "xr_rm_mujoco"], ) class LauncherCleanupTest(unittest.TestCase): def test_xrobotoolkit_cleanup_policy_only_stops_service_on_window_close(self) -> None: stop_all_patterns = set( launcher_ui._xrobotoolkit_cleanup_patterns(stop_pc_service=False) ) window_close_patterns = set( launcher_ui._xrobotoolkit_cleanup_patterns(stop_pc_service=True) ) self.assertLessEqual( {"RobotLinuxDemo.x86_64", "PXREAClientUnity"}, stop_all_patterns, ) self.assertNotIn("RoboticsServiceProcess", stop_all_patterns) self.assertIn("RoboticsServiceProcess", window_close_patterns) def test_close_and_stop_all_select_different_pc_service_policies(self) -> None: app = object.__new__(launcher_ui.LauncherApp) calls = [] class Root: destroyed = False def destroy(self) -> None: self.destroyed = True app.root = Root() app.stop_launched_processes = lambda **kwargs: calls.append(kwargs) or True app.kill_launched_processes() app.on_close_requested() self.assertEqual( calls, [ {"confirm": True, "notify": True, "stop_pc_service": False}, {"confirm": True, "notify": False, "stop_pc_service": True}, ], ) self.assertTrue(app.root.destroyed) def test_stop_all_keeps_oldest_pc_service_and_stops_duplicates(self) -> None: app = object.__new__(launcher_ui.LauncherApp) app.status = mock.Mock() app.close_related_terminal_windows = lambda: 0 def fake_check_output(command, **_kwargs): if command == ["pgrep", "-o", "-f", "RoboticsServiceProcess"]: return "101\n" if command == ["pgrep", "-f", "RoboticsServiceProcess"]: return "101\n202\n303\n" raise subprocess.CalledProcessError(1, command) with ( mock.patch.object( launcher_ui.subprocess, "check_output", side_effect=fake_check_output, ), mock.patch.object(launcher_ui.os, "kill") as kill, mock.patch.object(launcher_ui.time, "sleep"), ): app.stop_launched_processes( confirm=False, notify=False, stop_pc_service=False, ) self.assertEqual( kill.call_args_list, [ mock.call(202, signal.SIGTERM), mock.call(303, signal.SIGTERM), ], ) def test_stop_all_and_window_close_stop_mujoco(self) -> None: for stop_pc_service in (False, True): app = object.__new__(launcher_ui.LauncherApp) app.status = mock.Mock() app.close_related_terminal_windows = lambda: 0 def fake_check_output(command, **_kwargs): if command == ["pgrep", "-f", "dual_arm_simulator"]: return "404\n" raise subprocess.CalledProcessError(1, command) with ( mock.patch.object( launcher_ui.subprocess, "check_output", side_effect=fake_check_output, ), mock.patch.object(launcher_ui.os, "kill") as kill, mock.patch.object(launcher_ui.time, "sleep"), ): app.stop_launched_processes( confirm=False, notify=False, stop_pc_service=stop_pc_service, ) kill.assert_called_once_with(404, signal.SIGTERM) if __name__ == "__main__": unittest.main()