diff --git a/test/system/tools/fastdds/test_discovery_parser.py b/test/system/tools/fastdds/test_discovery_parser.py index 1cfb27a9c25..be61140ff8e 100644 --- a/test/system/tools/fastdds/test_discovery_parser.py +++ b/test/system/tools/fastdds/test_discovery_parser.py @@ -493,6 +493,245 @@ def test_parser_set_when_on(self, mock_rpc_stopall, mock_rpc_nbrequest, mock_rpc mock_rpc_brequest.assert_called_once() mock_exit.assert_not_called() + # @patch('parser.client_cli.run_request_nb') + # @patch('parser.client_cli.stop_all_request') + # @patch('parser.subprocess.run') + # @patch('parser.Path') + # @patch('parser.sys.exit') + # def test_parser_shutdown_command(self, mock_exit, mock_path, mock_subprocess_run): + # mock_path.return_value.resolve.return_value = Path('/fake/path') + # mock_subprocess_run.return_value = MagicMock(returncode=0, stdout='') + + # argv = ['shutdown'] + # with patch.object(sys, 'argv', argv): + # Parser(argv) + + # mock_exit.assert_not_called() + + # if command_int == command_to_int[Command.SHUTDOWN]: + # if not self.__is_daemon_running(): + # print('The Fast DDS daemon is not running.') + # raise SystemExit(0) + # print('After exit') + # print(client_cli.stop_all_request(get_sig_idx(signal.SIGTERM))) + # for p in Path(self.__shm_dir()).glob("*_servers.txt"): + # p.unlink() + # self.__stop_daemon() + # elif command_int == command_to_int[Command.AUTO] or command_int == command_to_int[Command.START]: + # self.__start_daemon(tool_path) + # self.__add_remote_servers_to_args(args_for_cpp) + # print(client_cli.run_request_nb(domain, args_for_cpp)) + # elif command_int == command_to_int[Command.STOP]: + # if not self.__is_daemon_running(): + # print('The Fast DDS daemon is not running.') + # raise SystemExit(0) + # print(client_cli.stop_request(domain, get_sig_idx(signal.SIGINT))) + # elif command_int == command_to_int[Command.ADD]: + # self.__start_daemon(tool_path) + # self.__add_remote_servers_to_args(args_for_cpp) + # print(client_cli.run_request_b(domain, args_for_cpp, True)) + # elif command_int == command_to_int[Command.SET]: + # self.__start_daemon(tool_path) + # print(client_cli.run_request_b(domain, args_for_cpp, True)) + # elif command_int == command_to_int[Command.LIST]: + # self.__start_daemon(tool_path) + # print(client_cli.run_request_b(domain, args_for_cpp, False)) + # pass + # elif command_int == command_to_int[Command.INFO]: + # print('Info mode not implemented yet.') + # pass + # else: + # print('Fast DDS CLI Error: Unknown command') + + # @patch('parser.os.path.exists') + # @patch('parser.Path') + # @patch('parser.sys.exit') + # def test_find_tool_path_posix(self, mock_exit, mock_path, mock_exists): + # mock_path.return_value.resolve.return_value = Path('/fake/path') + # mock_exists.side_effect = [False, True] + + # parser = Parser([]) + # tool_path = parser._Parser__find_tool_path() + + # self.assertEqual(tool_path, Path('/fake/path/../../../bin/fast-discovery-server')) + # mock_exit.assert_not_called() + + # @patch('parser.os.path.exists') + # @patch('parser.Path') + # @patch('parser.sys.exit') + # def test_find_tool_path_posix_not_installed(self, mock_exit, mock_path, mock_exists): + # mock_path.return_value.resolve.return_value = Path('/fake/path') + # mock_exists.side_effect = [False, False] + + # parser = Parser([]) + # with self.assertRaises(SystemExit): + # parser._Parser__find_tool_path() + + # mock_exit.assert_called_with(1) + + # @patch('parser.os.path.exists') + # @patch('parser.Path') + # @patch('parser.sys.exit') + # @patch('parser.os.name', 'nt') + # def test_find_tool_path_windows(self, mock_exit, mock_path, mock_exists): + # mock_path.return_value.resolve.return_value = Path('/fake/path') + # mock_exists.side_effect = [False, True] + + # parser = Parser([]) + # tool_path = parser._Parser__find_tool_path() + + # self.assertEqual(tool_path, Path('/fake/path/../../../bin/fast-discovery-server.exe')) + # mock_exit.assert_not_called() + + # @patch('parser.os.path.exists') + # @patch('parser.Path') + # @patch('parser.sys.exit') + # @patch('parser.os.name', 'nt') + # def test_find_tool_path_windows_bat(self, mock_exit, mock_path, mock_exists): + # mock_path.return_value.resolve.return_value = Path('/fake/path') + # mock_exists.side_effect = [False, False, True] + + # parser = Parser([]) + # tool_path = parser._Parser__find_tool_path() + + # self.assertEqual(tool_path, Path('/fake/path/../../../bin/fast-discovery-server.bat')) + # mock_exit.assert_not_called() + + # @patch('parser.os.path.exists') + # @patch('parser.Path') + # @patch('parser.sys.exit') + # @patch('parser.os.name', 'nt') + # def test_find_tool_path_windows_not_installed(self, mock_exit, mock_path, mock_exists): + # mock_path.return_value.resolve.return_value = Path('/fake/path') + # mock_exists.side_effect = [False, False, False] + + # parser = Parser([]) + # with self.assertRaises(SystemExit): + # parser._Parser__find_tool_path() + + # mock_exit.assert_called_with(1) + + # @patch('parser.os.path.exists') + # @patch('parser.Path') + # @patch('parser.sys.exit') + # @patch('parser.os.name', 'unsupported_os') + # def test_find_tool_path_unsupported_os(self, mock_exit, mock_path, mock_exists): + # mock_path.return_value.resolve.return_value = Path('/fake/path') + + # parser = Parser([]) + # with self.assertRaises(SystemExit): + # parser._Parser__find_tool_path() + + # mock_exit.assert_called_with(1) + + # @patch('parser.re.sub') + # def test_edit_tool_text(self, mock_re_sub): + # mock_re_sub.return_value = 'fastdds discovery -h' + # parser = Parser([]) + # result = parser._Parser__edit_tool_text('Usage: fake-tool -h', 'fake-tool') + # self.assertEqual(result, 'fastdds discovery -h') + # mock_re_sub.assert_called_with('fake-tool', 'fastdds discovery', 'Usage: fake-tool -h') + + # @patch('parser.spawn_daemon') + # @patch('parser.print') + # def test_start_daemon(self, mock_print, mock_spawn_daemon): + # mock_spawn_daemon.return_value = True + # parser = Parser([]) + # parser._Parser__start_daemon('/fake/path') + # mock_print.assert_called_with('The Fast DDS daemon has been started.') + + # mock_spawn_daemon.return_value = False + # parser._Parser__start_daemon('/fake/path') + # mock_print.assert_called_with('The Fast DDS daemon is already running.') + + # @patch('parser.shutdown_daemon') + # @patch('parser.print') + # def test_stop_daemon(self, mock_print, mock_shutdown_daemon): + # mock_shutdown_daemon.return_value = True + # parser = Parser([]) + # parser._Parser__stop_daemon() + # mock_print.assert_called_with('The Fast DDS daemon has been stopped.') + + # mock_shutdown_daemon.return_value = False + # parser._Parser__stop_daemon() + # mock_print.assert_called_with('The Fast DDS daemon is not running.') + + # @patch('parser.is_daemon_running') + # def test_is_daemon_running(self, mock_is_daemon_running): + # mock_is_daemon_running.return_value = True + # parser = Parser([]) + # result = parser._Parser__is_daemon_running() + # self.assertTrue(result) + + # mock_is_daemon_running.return_value = False + # result = parser._Parser__is_daemon_running() + # self.assertFalse(result) + + # @patch('parser.os.getenv') + # @patch('parser.print') + # def test_get_domain_id_from_env(self, mock_print, mock_getenv): + # mock_getenv.return_value = '10' + # parser = Parser([]) + # result = parser._Parser__get_domain_id_from_env() + # self.assertEqual(result, 10) + + # mock_getenv.return_value = '300' + # result = parser._Parser__get_domain_id_from_env() + # self.assertEqual(result, 0) + # mock_print.assert_called_with("Error: Found invalid Domain ID (300) in environment variable.") + + # mock_getenv.return_value = None + # result = parser._Parser__get_domain_id_from_env() + # self.assertEqual(result, 0) + + # @patch('parser.os.getenv') + # def test_get_remote_servers_from_env(self, mock_getenv): + # mock_getenv.return_value = 'server1,server2' + # parser = Parser([]) + # result = parser._Parser__get_remote_servers_from_env() + # self.assertEqual(result, 'server1,server2') + + # mock_getenv.return_value = None + # result = parser._Parser__get_remote_servers_from_env() + # self.assertEqual(result, '') + + # @patch('parser.Parser._Parser__get_remote_servers_from_env') + # def test_add_remote_servers_to_args(self, mock_get_remote_servers_from_env): + # mock_get_remote_servers_from_env.return_value = 'server1,server2' + # parser = Parser([]) + # args = [] + # parser._Parser__add_remote_servers_to_args(args) + # self.assertIn('server1,server2', args) + + # mock_get_remote_servers_from_env.return_value = '' + # args = [] + # parser._Parser__add_remote_servers_to_args(args) + # self.assertNotIn('server1,server2', args) + + # @patch('parser.platform.mac_ver') + # @patch('parser.Path') + # def test_shm_dir_mac(self, mock_path, mock_mac_ver): + # mock_mac_ver.return_value = ('10.15.7', ('', '', ''), 'x86_64') + # mock_path.return_value.resolve.return_value = Path('/private/tmp/boost_interprocess/') + # parser = Parser([]) + # result = parser._Parser__shm_dir() + # self.assertEqual(result, Path('/private/tmp/boost_interprocess/')) + + # @patch('parser.platform.mac_ver') + # @patch('parser.Path') + # def test_shm_dir_linux(self, mock_path, mock_mac_ver): + # mock_mac_ver.return_value = ('', ('', '', ''), '') + # mock_path.return_value.resolve.return_value = Path('/dev/shm/') + # parser = Parser([]) + # result = parser._Parser__shm_dir() + # self.assertEqual(result, Path('/dev/shm/')) + + # @patch('parser.os.name', 'unsupported_os') + # def test_shm_dir_unsupported_os(self): + # parser = Parser([]) + # with self.assertRaises(RuntimeError): + # parser._Parser__shm_dir() + if __name__ == '__main__': unittest.main() \ No newline at end of file