Coverage for virtuoso_utilities / isql_helpers.py: 48%

1# SPDX-FileCopyrightText: 2025 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5""" 

6Helper functions for executing ISQL commands and scripts against Virtuoso, 

7handling both direct execution and execution via Docker. 

8""" 

9import argparse 

10import os 

11import shlex 

12import subprocess 

13import sys 

14from typing import Union 

15 

16 

17def _run_subprocess( 

18 command: Union[list[str], str], 

19 use_shell: bool = False, 

20 encoding: str = 'utf-8' 

21) -> tuple[int, str, str]: 

22 """Internal helper to run a subprocess command. Always captures output.""" 

23 try: 

24 process = subprocess.run( 

25 command, 

26 shell=use_shell, 

27 capture_output=True, 

28 text=True, 

29 check=False, 

30 encoding=encoding 

31 ) 

32 stdout = process.stdout.strip() if process.stdout else "" 

33 stderr = process.stderr.strip() if process.stderr else "" 

34 return process.returncode, stdout, stderr 

35 except Exception as e: 

36 print(f"Subprocess execution failed: {e}", file=sys.stderr) 

37 print(f"Command: {command}", file=sys.stderr) 

38 return -1, "", str(e) 

39 

40def run_isql_command( 

41 args: argparse.Namespace, 

42 sql_command: Union[str, None] = None, 

43 script_path: Union[str, None] = None, 

44 ignore_errors: bool = False, 

45) -> tuple[bool, str, str]: 

46 """ 

47 Executes a SQL command or script using the 'isql' utility, either directly 

48 or via 'docker exec'. Output is always captured and only shown on error. 

49 

50 Exactly one of `sql_command` or `script_path` must be provided. 

51 

52 Args: 

53 args (argparse.Namespace): Parsed command-line arguments containing 

54 connection details and paths. Must have 

55 attributes: docker_container, host, port, 

56 user, password. If docker_container 

57 is set, must also have docker_path and 

58 docker_isql_path. Otherwise, must have 

59 isql_path. 

60 sql_command (Union[str, None]): The SQL command string to execute. 

61 script_path (Union[str, None]): The path to the SQL script file to execute. 

62 ignore_errors (bool): If True, print errors but return True anyway. 

63 

64 Returns: 

65 tuple: (success_status, stdout, stderr) 

66 success_status is True if the command/script ran without error 

67 (exit code 0) or if ignore_errors is True. 

68 stdout and stderr contain the respective outputs. 

69 

70 Raises: 

71 ValueError: If neither or both sql_command and script_path are provided. 

72 """ 

73 if not ((sql_command is None) ^ (script_path is None)): 

74 raise ValueError("Exactly one of sql_command or script_path must be provided.") 

75 

76 command_to_run: Union[list[str], str] = [] 

77 use_shell = False 

78 effective_isql_path_for_error = "" 

79 command_description = "" 

80 

81 if args.docker_container: 

82 if not hasattr(args, 'docker_path') or not args.docker_path: 

83 print("Error: 'docker_path' argument missing for Docker execution.", file=sys.stderr) 

84 return False, "", "'docker_path' argument missing" 

85 if not hasattr(args, 'docker_isql_path') or not args.docker_isql_path: 

86 print("Error: 'docker_isql_path' argument missing for Docker execution.", file=sys.stderr) 

87 return False, "", "'docker_isql_path' argument missing" 

88 

89 effective_isql_path_for_error = f"'{args.docker_isql_path}' inside container '{args.docker_container}' via '{args.docker_path}'" 

90 

91 exec_content = "" 

92 if sql_command: 

93 exec_content = sql_command 

94 command_description = "ISQL command (Docker)" 

95 else: 

96 command_description = "ISQL script (Docker)" 

97 if not os.path.exists(script_path): 

98 print(f"Error: Script file not found at '{script_path}'", file=sys.stderr) 

99 return False, "", f"Script file not found: {script_path}" 

100 print("Reading script content for docker exec...", file=sys.stderr) 

101 try: 

102 with open(script_path, 'r', encoding='utf-8') as f: 

103 sql_content = f.read() 

104 exec_content = sql_content.replace('\n', ' ').strip() 

105 except Exception as e: 

106 print(f"Error reading SQL script file '{script_path}': {e}", file=sys.stderr) 

107 return False, "", str(e) 

108 

109 docker_internal_host = "localhost" 

110 docker_internal_port = 1111 

111 command_to_run = [ 

112 args.docker_path, 

113 'exec', 

114 args.docker_container, 

115 args.docker_isql_path, 

116 f"{docker_internal_host}:{docker_internal_port}", 

117 args.user, 

118 args.password, 

119 f"exec={exec_content}" 

120 ] 

121 

122 else: 

123 if not hasattr(args, 'isql_path') or not args.isql_path: 

124 print("Error: 'isql_path' argument missing for non-Docker execution.", file=sys.stderr) 

125 return False, "", "'isql_path' argument missing" 

126 

127 effective_isql_path_for_error = f"'{args.isql_path}' on host" 

128 

129 if sql_command: 

130 command_description = "ISQL command (Local)" 

131 command_to_run = [ 

132 args.isql_path, 

133 f"{args.host}:{args.port}", 

134 args.user, 

135 args.password, 

136 f"exec={sql_command}" 

137 ] 

138 else: 

139 command_description = "ISQL script (Local)" 

140 if not os.path.exists(script_path): 

141 print(f"Error: Script file not found at '{script_path}'", file=sys.stderr) 

142 return False, "", f"Script file not found: {script_path}" 

143 

144 effective_isql_path_for_error += " using shell redirection" 

145 use_shell = True 

146 

147 base_command_list = [ 

148 args.isql_path, 

149 f"{args.host}:{args.port}", 

150 args.user, 

151 args.password, 

152 f"< {shlex.quote(script_path)}" # Use shlex.quote for safety 

153 ] 

154 command_to_run = " ".join(base_command_list) 

155 

156 try: 

157 returncode, stdout, stderr = _run_subprocess(command_to_run, use_shell=use_shell) 

158 

159 if returncode != 0: 

160 # Handle specific FileNotFoundError after subprocess call 

161 if ("No such file or directory" in stderr or "not found" in stderr or returncode == 127): 

162 # Distinguish between primary executable not found vs other issues 

163 missing_cmd = args.docker_path if args.docker_container else args.isql_path 

164 print(f"Error: Command '{missing_cmd}' or related component not found.", file=sys.stderr) 

165 if args.docker_container: 

166 print(f"Make sure '{args.docker_path}' is installed and in your PATH, and the container/isql path is correct.", file=sys.stderr) 

167 else: 

168 print(f"Make sure Virtuoso client tools (containing '{args.isql_path}') are installed and in your PATH.", file=sys.stderr) 

169 if use_shell: 

170 print(f"Check shell environment if using local script execution.", file=sys.stderr) 

171 return False, stdout, f"Executable or shell component not found: {missing_cmd}" 

172 

173 return ignore_errors, stdout, stderr 

174 return True, stdout, stderr 

175 except Exception as e: 

176 # Catch unexpected errors *around* the subprocess call if any 

177 print(f"An unexpected error occurred preparing or handling {command_description}: {e}", file=sys.stderr) 

178 print(f"Command context: {command_to_run}", file=sys.stderr) 

179 return False, "", str(e)