WIP

Signed-off-by: cferreiragonz <[email protected]>

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()