Coverage for mcpgateway / utils / error_formatter.py: 100%

1# -*- coding: utf-8 -*- 

2"""Location: ./mcpgateway/utils/error_formatter.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Mihai Criveti 

6 

7ContextForge Centralized for Pydantic validation error, SQL exception. 

8This module provides centralized error formatting for ContextForge, 

9transforming technical Pydantic validation errors and SQLAlchemy database 

10exceptions into user-friendly messages suitable for API responses. 

11 

12The ErrorFormatter class handles: 

13- Pydantic ValidationError formatting 

14- SQLAlchemy DatabaseError and IntegrityError formatting 

15- Mapping technical error messages to user-friendly explanations 

16- Consistent error response structure 

17 

18Examples: 

19 >>> from mcpgateway.utils.error_formatter import ErrorFormatter 

20 >>> from pydantic import ValidationError 

21 >>> 

22 >>> # Format validation errors 

23 >>> formatter = ErrorFormatter() 

24 >>> # formatted_error = formatter.format_validation_error(validation_error) 

25""" 

26 

27# Standard 

28from typing import Any, Dict 

29 

30# Third-Party 

31from pydantic import ValidationError 

32from sqlalchemy.exc import DatabaseError, IntegrityError 

33 

34# First-Party 

35from mcpgateway.services.logging_service import LoggingService 

36 

37# Initialize logging service first 

38logging_service = LoggingService() 

39logger = logging_service.get_logger(__name__) 

40 

41 

42class ErrorFormatter: 

43 """Transform technical errors into user-friendly messages. 

44 

45 Provides static methods to convert Pydantic validation errors and 

46 SQLAlchemy database exceptions into consistent, user-friendly error 

47 responses suitable for API consumption. 

48 

49 Examples: 

50 >>> formatter = ErrorFormatter() 

51 >>> isinstance(formatter, ErrorFormatter) 

52 True 

53 """ 

54 

55 @staticmethod 

56 def format_validation_error(error: ValidationError) -> Dict[str, Any]: 

57 """Convert Pydantic errors to user-friendly format. 

58 

59 Transforms Pydantic ValidationError objects into a structured 

60 dictionary containing user-friendly error messages. Maps technical 

61 validation messages to more understandable explanations. 

62 

63 Args: 

64 error (ValidationError): The Pydantic validation error to format 

65 

66 Returns: 

67 Dict[str, Any]: A dictionary with formatted error details containing: 

68 - message: General error description 

69 - details: List of field-specific errors 

70 - success: Always False for errors 

71 

72 Examples: 

73 >>> from pydantic import BaseModel, ValidationError, field_validator 

74 >>> # Create a test model with validation 

75 >>> class TestModel(BaseModel): 

76 ... name: str 

77 ... @field_validator('name') 

78 ... def validate_name(cls, v): 

79 ... if not v.startswith('A'): 

80 ... raise ValueError('Tool name must start with a letter, number, or underscore') 

81 ... return v 

82 >>> # Test validation error formatting 

83 >>> try: 

84 ... TestModel(name="B123") 

85 ... except ValidationError as e: 

86 ... print(type(e)) 

87 ... result = ErrorFormatter.format_validation_error(e) 

88 <class 'pydantic_core._pydantic_core.ValidationError'> 

89 >>> result['message'] 

90 'Validation failed: Name must start with a letter, number, or underscore and contain only letters, numbers, periods, underscores, hyphens, and slashes' 

91 >>> result['success'] 

92 False 

93 >>> len(result['details']) > 0 

94 True 

95 >>> result['details'][0]['field'] 

96 'name' 

97 >>> 'must start with a letter, number, or underscore' in result['details'][0]['message'] 

98 True 

99 

100 >>> # Test with multiple errors 

101 >>> class MultiFieldModel(BaseModel): 

102 ... name: str 

103 ... url: str 

104 ... @field_validator('name') 

105 ... def validate_name(cls, v): 

106 ... if len(v) > 255: 

107 ... raise ValueError('Tool name exceeds maximum length') 

108 ... return v 

109 ... @field_validator('url') 

110 ... def validate_url(cls, v): 

111 ... if not v.startswith('http'): 

112 ... raise ValueError('Tool URL must start with http') 

113 ... return v 

114 >>> 

115 >>> try: 

116 ... MultiFieldModel(name='A' * 300, url='ftp://invalid') 

117 ... except ValidationError as e: 

118 ... print(type(e)) 

119 ... result = ErrorFormatter.format_validation_error(e) 

120 <class 'pydantic_core._pydantic_core.ValidationError'> 

121 >>> len(result['details']) 

122 2 

123 >>> any('too long' in detail['message'] for detail in result['details']) 

124 True 

125 >>> any('valid HTTP' in detail['message'] for detail in result['details']) 

126 True 

127 """ 

128 errors = [] 

129 

130 for err in error.errors(): 

131 loc = err.get("loc", ["field"]) 

132 field = loc[-1] if loc else "field" 

133 msg = err.get("msg", "Invalid value") 

134 

135 # Map technical messages to user-friendly ones 

136 user_message = ErrorFormatter._get_user_message(field, msg) 

137 errors.append({"field": field, "message": user_message}) 

138 

139 # Log the full error for debugging 

140 logger.debug(f"Validation error: {error}") 

141 

142 return {"message": f"Validation failed: {user_message}", "details": errors, "success": False} 

143 

144 @staticmethod 

145 def _get_user_message(field: str, technical_msg: str) -> str: 

146 """Map technical validation messages to user-friendly ones. 

147 

148 Converts technical validation error messages into user-friendly 

149 explanations based on pattern matching. Provides field-specific 

150 context in the returned message. 

151 

152 Args: 

153 field (str): The field name that failed validation 

154 technical_msg (str): The technical validation message from Pydantic 

155 

156 Returns: 

157 str: User-friendly error message with field context 

158 

159 Examples: 

160 >>> # Test letter requirement mapping 

161 >>> msg = ErrorFormatter._get_user_message("name", "Tool name must start with a letter, number, or underscore") 

162 >>> msg 

163 'Name must start with a letter, number, or underscore and contain only letters, numbers, periods, underscores, hyphens, and slashes' 

164 

165 >>> # Test length validation mapping 

166 >>> msg = ErrorFormatter._get_user_message("description", "Tool name exceeds maximum length") 

167 >>> msg 

168 'Description is too long (maximum 255 characters)' 

169 

170 >>> # Test URL validation mapping 

171 >>> msg = ErrorFormatter._get_user_message("endpoint", "Tool URL must start with http") 

172 >>> msg 

173 'Endpoint must be a valid HTTP or WebSocket URL' 

174 

175 >>> # Test directory traversal validation 

176 >>> msg = ErrorFormatter._get_user_message("path", "cannot contain directory traversal") 

177 >>> msg 

178 'Path contains invalid characters' 

179 

180 >>> # Test HTML injection validation 

181 >>> msg = ErrorFormatter._get_user_message("content", "contains HTML tags") 

182 >>> msg 

183 'Content cannot contain HTML or script tags' 

184 

185 >>> # Test fallback for unknown messages 

186 >>> msg = ErrorFormatter._get_user_message("custom_field", "Some unknown error") 

187 >>> msg 

188 'Invalid custom_field' 

189 """ 

190 mappings = { 

191 "Tool name must start with a letter, number, or underscore": f"{field.title()} must start with a letter, number, or underscore and contain only letters, numbers, periods, underscores, hyphens, and slashes", 

192 "Tool name exceeds maximum length": f"{field.title()} is too long (maximum 255 characters)", 

193 "Tool URL must start with": f"{field.title()} must be a valid HTTP or WebSocket URL", 

194 "cannot contain directory traversal": f"{field.title()} contains invalid characters", 

195 "contains HTML tags": f"{field.title()} cannot contain HTML or script tags", 

196 "Server ID must be a valid UUID format": f"{field.title()} must be a valid UUID", 

197 } 

198 

199 for pattern, friendly_msg in mappings.items(): 

200 if pattern in technical_msg: 

201 return friendly_msg 

202 

203 # Default fallback 

204 return f"Invalid {field}" 

205 

206 @staticmethod 

207 def format_database_error(error: DatabaseError) -> Dict[str, Any]: 

208 """Convert database errors to user-friendly format. 

209 

210 Transforms SQLAlchemy database exceptions into structured error 

211 responses. Handles common integrity constraint violations and 

212 provides specific messages for known error patterns. 

213 

214 Args: 

215 error (DatabaseError): The SQLAlchemy database error to format 

216 

217 Returns: 

218 Dict[str, Any]: A dictionary with formatted error details containing: 

219 - message: User-friendly error description 

220 - success: Always False for errors 

221 

222 Examples: 

223 >>> from unittest.mock import Mock 

224 >>> 

225 >>> # Test UNIQUE constraint on gateway URL 

226 >>> mock_error = Mock(spec=IntegrityError) 

227 >>> mock_error.orig = Mock() 

228 >>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: gateways.url" 

229 >>> result = ErrorFormatter.format_database_error(mock_error) 

230 >>> result['message'] 

231 'A gateway with this URL already exists' 

232 >>> result['success'] 

233 False 

234 

235 >>> # Test UNIQUE constraint on gateway slug 

236 >>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: gateways.slug" 

237 >>> result = ErrorFormatter.format_database_error(mock_error) 

238 >>> result['message'] 

239 'A gateway with this name already exists' 

240 

241 >>> # Test UNIQUE constraint on tool name 

242 >>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: tools.name" 

243 >>> result = ErrorFormatter.format_database_error(mock_error) 

244 >>> result['message'] 

245 'A tool with this name already exists' 

246 

247 >>> # Test UNIQUE constraint on resource URI 

248 >>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: resources.uri" 

249 >>> result = ErrorFormatter.format_database_error(mock_error) 

250 >>> result['message'] 

251 'A resource with this URI already exists' 

252 

253 >>> # Test UNIQUE constraint on server name 

254 >>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: servers.name" 

255 >>> result = ErrorFormatter.format_database_error(mock_error) 

256 >>> result['message'] 

257 'A server with this name already exists' 

258 

259 >>> # Test UNIQUE constraint on prompt name 

260 >>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: prompts.name" 

261 >>> result = ErrorFormatter.format_database_error(mock_error) 

262 >>> result['message'] 

263 'A prompt with this name already exists' 

264 

265 >>> # Test FOREIGN KEY constraint 

266 >>> mock_error.orig.__str__ = lambda self: "FOREIGN KEY constraint failed" 

267 >>> result = ErrorFormatter.format_database_error(mock_error) 

268 >>> result['message'] 

269 'Referenced item not found' 

270 

271 >>> # Test NOT NULL constraint 

272 >>> mock_error.orig.__str__ = lambda self: "NOT NULL constraint failed" 

273 >>> result = ErrorFormatter.format_database_error(mock_error) 

274 >>> result['message'] 

275 'Required field is missing' 

276 

277 >>> # Test CHECK constraint 

278 >>> mock_error.orig.__str__ = lambda self: "CHECK constraint failed: invalid_data" 

279 >>> result = ErrorFormatter.format_database_error(mock_error) 

280 >>> result['message'] 

281 'Validation failed. Please check the input data.' 

282 

283 >>> # Test generic database error 

284 >>> generic_error = Mock(spec=DatabaseError) 

285 >>> generic_error.orig = None 

286 >>> result = ErrorFormatter.format_database_error(generic_error) 

287 >>> result['message'] 

288 'Unable to complete the operation. Please try again.' 

289 >>> result['success'] 

290 False 

291 """ 

292 error_str = str(error.orig) if hasattr(error, "orig") else str(error) 

293 

294 # Log full error 

295 logger.error(f"Database error: {error}") 

296 

297 # Map common database errors 

298 if isinstance(error, IntegrityError): 

299 # Token name uniqueness: check before generic UNIQUE handler so the specific message 

300 # takes priority. PostgreSQL reports the constraint name (either the db.py name or the 

301 # Alembic migration name); SQLite reports the column paths. 

302 if ( 

303 "uq_email_api_tokens_user_name_team" in error_str 

304 or "uq_email_api_tokens_user_name" in error_str 

305 or "uq_email_api_tokens_user_email_name" in error_str 

306 or ("email_api_tokens.user_email" in error_str and "email_api_tokens.name" in error_str) 

307 ): 

308 return { 

309 "message": "A token with this name already exists for this user in the same team scope. Token names must be unique per user per team. Please choose a different name.", 

310 "success": False, 

311 } 

312 if "UNIQUE constraint failed" in error_str: 

313 if "gateways.url" in error_str: 

314 return {"message": "A gateway with this URL already exists", "success": False} 

315 elif "gateways.slug" in error_str: 

316 return {"message": "A gateway with this name already exists", "success": False} 

317 elif "tools.name" in error_str: 

318 return {"message": "A tool with this name already exists", "success": False} 

319 elif "resources.uri" in error_str: 

320 return {"message": "A resource with this URI already exists", "success": False} 

321 elif "servers.name" in error_str: 

322 return {"message": "A server with this name already exists", "success": False} 

323 elif "prompts.name" in error_str: 

324 return {"message": "A prompt with this name already exists", "success": False} 

325 elif "servers.id" in error_str: 

326 return {"message": "A server with this ID already exists", "success": False} 

327 elif "a2a_agents.slug" in error_str: 

328 return {"message": "An A2A agent with this name already exists", "success": False} 

329 

330 elif "FOREIGN KEY constraint failed" in error_str: 

331 return {"message": "Referenced item not found", "success": False} 

332 elif "NOT NULL constraint failed" in error_str: 

333 return {"message": "Required field is missing", "success": False} 

334 elif "CHECK constraint failed:" in error_str: 

335 return {"message": "Validation failed. Please check the input data.", "success": False} 

336 

337 # Generic database error 

338 return {"message": "Unable to complete the operation. Please try again.", "success": False}