Coverage for fastblocks / mcp / health.py: 72%

1"""Health check and validation system for FastBlocks adapters.""" 

2 

3import asyncio 

4import time 

5from datetime import datetime 

6from typing import Any 

7 

8from .registry import AdapterRegistry 

9 

10 

11class HealthCheckResult: 

12 """Result of a health check operation.""" 

13 

14 def __init__( 

15 self, 

16 adapter_name: str, 

17 status: str, 

18 message: str = "", 

19 details: dict[str, Any] | None = None, 

20 duration_ms: float = 0.0, 

21 timestamp: datetime | None = None, 

22 ): 

23 self.adapter_name = adapter_name 

24 self.status = status # 'healthy', 'warning', 'error', 'unknown' 

25 self.message = message 

26 self.details = details or {} 

27 self.duration_ms = duration_ms 

28 self.timestamp = timestamp or datetime.now() 

29 

30 def to_dict(self) -> dict[str, Any]: 

31 """Convert result to dictionary.""" 

32 return { 

33 "adapter_name": self.adapter_name, 

34 "status": self.status, 

35 "message": self.message, 

36 "details": self.details, 

37 "duration_ms": self.duration_ms, 

38 "timestamp": self.timestamp.isoformat(), 

39 } 

40 

41 

42class HealthCheckSystem: 

43 """Health monitoring and validation system for adapters.""" 

44 

45 def __init__(self, registry: AdapterRegistry): 

46 """Initialize health check system.""" 

47 self.registry = registry 

48 self._check_history: dict[str, list[HealthCheckResult]] = {} 

49 self._check_config: dict[str, dict[str, Any]] = {} 

50 self._running_checks: dict[str, bool] = {} 

51 

52 async def check_adapter_health(self, adapter_name: str) -> HealthCheckResult: 

53 """Perform comprehensive health check on an adapter.""" 

54 start_time = time.time() 

55 

56 try: 

57 # Prevent concurrent checks on same adapter 

58 if self._running_checks.get(adapter_name, False): 

59 return HealthCheckResult( 

60 adapter_name, "warning", "Health check already in progress" 

61 ) 

62 

63 self._running_checks[adapter_name] = True 

64 

65 # Basic validation check 

66 validation_result = await self.registry.validate_adapter(adapter_name) 

67 

68 if not validation_result["valid"]: 

69 duration_ms = (time.time() - start_time) * 1000 

70 result = HealthCheckResult( 

71 adapter_name, 

72 "error", 

73 f"Validation failed: {', '.join(validation_result['errors'])}", 

74 validation_result, 

75 duration_ms, 

76 ) 

77 else: 

78 # Get adapter instance for functional tests 

79 adapter = await self.registry.get_adapter(adapter_name) 

80 

81 if not adapter: 

82 duration_ms = (time.time() - start_time) * 1000 

83 result = HealthCheckResult( 

84 adapter_name, 

85 "error", 

86 "Could not instantiate adapter", 

87 duration_ms=duration_ms, 

88 ) 

89 else: 

90 # Perform functional health checks 

91 result = await self._perform_functional_checks( 

92 adapter_name, adapter, start_time 

93 ) 

94 

95 except Exception as e: 

96 duration_ms = (time.time() - start_time) * 1000 

97 result = HealthCheckResult( 

98 adapter_name, 

99 "error", 

100 f"Health check failed: {e}", 

101 duration_ms=duration_ms, 

102 ) 

103 finally: 

104 self._running_checks[adapter_name] = False 

105 

106 # Store result in history 

107 self._store_check_result(result) 

108 

109 return result 

110 

111 def _determine_health_status( 

112 self, checks: list[str], warnings: list[str] 

113 ) -> tuple[str, str]: 

114 """Determine health status and message based on checks and warnings.""" 

115 if checks: 

116 status = "warning" if warnings else "healthy" 

117 message = f"Passed {len(checks)} checks" 

118 if warnings: 

119 message += f", {len(warnings)} warnings" 

120 else: 

121 status = "error" 

122 message = "No functional checks passed" 

123 return status, message 

124 

125 async def _perform_category_specific_checks( 

126 self, adapter_info: Any, adapter: Any 

127 ) -> list[str]: 

128 """Perform category-specific health checks.""" 

129 checks = [] 

130 category = adapter_info.category 

131 

132 # Category-specific health checks 

133 if category == "images": 

134 checks.extend(await self._check_image_adapter(adapter)) 

135 elif category == "styles": 

136 checks.extend(await self._check_style_adapter(adapter)) 

137 elif category == "icons": 

138 checks.extend(await self._check_icon_adapter(adapter)) 

139 elif category == "fonts": 

140 checks.extend(await self._check_font_adapter(adapter)) 

141 elif category == "templates": 

142 checks.extend(await self._check_template_adapter(adapter)) 

143 

144 return checks 

145 

146 async def _check_acb_registration( 

147 self, adapter_name: str 

148 ) -> tuple[list[str], list[str]]: 

149 """Check ACB registration for the adapter.""" 

150 checks = [] 

151 warnings = [] 

152 

153 try: 

154 from acb.depends import depends 

155 

156 registered_adapter = depends.get_sync(adapter_name) 

157 if registered_adapter: 

158 checks.append("Registered with ACB") 

159 else: 

160 warnings.append("Not registered with ACB") 

161 except Exception: 

162 warnings.append("ACB registration check failed") 

163 

164 return checks, warnings 

165 

166 async def _perform_functional_checks( 

167 self, adapter_name: str, adapter: Any, start_time: float 

168 ) -> HealthCheckResult: 

169 """Perform functional health checks on an adapter.""" 

170 checks = [] 

171 warnings = [] 

172 

173 # Check if adapter has required methods based on its type 

174 adapter_info = await self.registry.get_adapter_info(adapter_name) 

175 

176 if adapter_info: 

177 checks.extend( 

178 await self._perform_category_specific_checks(adapter_info, adapter) 

179 ) 

180 

181 # Check settings availability 

182 if hasattr(adapter, "settings"): 

183 checks.append("Settings available") 

184 else: 

185 warnings.append("No settings found") 

186 

187 # Check ACB registration 

188 acb_checks, acb_warnings = await self._check_acb_registration(adapter_name) 

189 checks.extend(acb_checks) 

190 warnings.extend(acb_warnings) 

191 

192 duration_ms = (time.time() - start_time) * 1000 

193 

194 # Determine overall status 

195 status, message = self._determine_health_status(checks, warnings) 

196 

197 return HealthCheckResult( 

198 adapter_name, 

199 status, 

200 message, 

201 { 

202 "checks_passed": checks, 

203 "warnings": warnings, 

204 "category": adapter_info.category if adapter_info else "unknown", 

205 }, 

206 duration_ms, 

207 ) 

208 

209 async def _check_image_adapter(self, adapter: Any) -> list[str]: 

210 """Check image adapter functionality.""" 

211 checks = [] 

212 

213 if hasattr(adapter, "get_img_tag"): 

214 checks.append("get_img_tag method available") 

215 

216 if hasattr(adapter, "get_image_url"): 

217 checks.append("get_image_url method available") 

218 

219 if hasattr(adapter, "upload_image"): 

220 checks.append("upload_image method available") 

221 

222 return checks 

223 

224 async def _check_style_adapter(self, adapter: Any) -> list[str]: 

225 """Check style adapter functionality.""" 

226 checks = [] 

227 

228 if hasattr(adapter, "get_stylesheet_links"): 

229 checks.append("get_stylesheet_links method available") 

230 

231 if hasattr(adapter, "get_component_class"): 

232 checks.append("get_component_class method available") 

233 

234 if hasattr(adapter, "get_utility_classes"): 

235 checks.append("get_utility_classes method available") 

236 

237 return checks 

238 

239 async def _check_icon_adapter(self, adapter: Any) -> list[str]: 

240 """Check icon adapter functionality.""" 

241 checks = [] 

242 

243 if hasattr(adapter, "get_icon_tag"): 

244 checks.append("get_icon_tag method available") 

245 

246 if hasattr(adapter, "get_stylesheet_links"): 

247 checks.append("get_stylesheet_links method available") 

248 

249 if hasattr(adapter, "get_icon_class"): 

250 checks.append("get_icon_class method available") 

251 

252 return checks 

253 

254 async def _check_font_adapter(self, adapter: Any) -> list[str]: 

255 """Check font adapter functionality.""" 

256 checks = [] 

257 

258 if hasattr(adapter, "get_font_import"): 

259 checks.append("get_font_import method available") 

260 

261 if hasattr(adapter, "get_font_family"): 

262 checks.append("get_font_family method available") 

263 

264 return checks 

265 

266 async def _check_template_adapter(self, adapter: Any) -> list[str]: 

267 """Check template adapter functionality.""" 

268 checks = [] 

269 

270 if hasattr(adapter, "render_template"): 

271 checks.append("render_template method available") 

272 

273 if hasattr(adapter, "get_template"): 

274 checks.append("get_template method available") 

275 

276 if hasattr(adapter, "list_templates"): 

277 checks.append("list_templates method available") 

278 

279 return checks 

280 

281 async def check_all_adapters(self) -> dict[str, HealthCheckResult]: 

282 """Perform health checks on all available adapters.""" 

283 available_adapters = await self.registry.list_available_adapters() 

284 results = {} 

285 

286 # Run checks concurrently 

287 tasks = [] 

288 for adapter_name in available_adapters.keys(): 

289 task = asyncio.create_task(self.check_adapter_health(adapter_name)) 

290 tasks.append((adapter_name, task)) 

291 

292 for adapter_name, task in tasks: 

293 try: 

294 result = await task 

295 results[adapter_name] = result 

296 except Exception as e: 

297 results[adapter_name] = HealthCheckResult( 

298 adapter_name, "error", f"Check failed: {e}" 

299 ) 

300 

301 return results 

302 

303 def _store_check_result(self, result: HealthCheckResult) -> None: 

304 """Store health check result in history.""" 

305 if result.adapter_name not in self._check_history: 

306 self._check_history[result.adapter_name] = [] 

307 

308 self._check_history[result.adapter_name].append(result) 

309 

310 # Keep only last 100 results per adapter 

311 if len(self._check_history[result.adapter_name]) > 100: 

312 self._check_history[result.adapter_name] = self._check_history[ 

313 result.adapter_name 

314 ][-100:] 

315 

316 def get_check_history( 

317 self, adapter_name: str, limit: int = 10 

318 ) -> list[HealthCheckResult]: 

319 """Get health check history for an adapter.""" 

320 return self._check_history.get(adapter_name, [])[-limit:] 

321 

322 def get_system_health_summary(self) -> dict[str, Any]: 

323 """Get overall system health summary.""" 

324 summary: dict[str, Any] = { 

325 "healthy_adapters": 0, 

326 "warning_adapters": 0, 

327 "error_adapters": 0, 

328 "unknown_adapters": 0, 

329 "total_adapters": 0, 

330 "last_check_time": None, 

331 "adapter_status": {}, 

332 } 

333 

334 latest_results = {} 

335 

336 # Get latest result for each adapter 

337 for adapter_name, history in self._check_history.items(): 

338 if history: 

339 latest_results[adapter_name] = history[-1] 

340 

341 # Count statuses 

342 for adapter_name, result in latest_results.items(): 

343 summary["adapter_status"][adapter_name] = { 

344 "status": result.status, 

345 "last_check": result.timestamp.isoformat(), 

346 "message": result.message, 

347 } 

348 

349 if result.status == "healthy": 

350 summary["healthy_adapters"] += 1 

351 elif result.status == "warning": 

352 summary["warning_adapters"] += 1 

353 elif result.status == "error": 

354 summary["error_adapters"] += 1 

355 else: 

356 summary["unknown_adapters"] += 1 

357 

358 summary["total_adapters"] = len(latest_results) 

359 

360 if latest_results: 

361 latest_time = max(result.timestamp for result in latest_results.values()) 

362 summary["last_check_time"] = latest_time.isoformat() 

363 

364 return summary 

365 

366 def configure_health_checks( 

367 self, adapter_name: str, config: dict[str, Any] 

368 ) -> None: 

369 """Configure health check parameters for an adapter.""" 

370 self._check_config[adapter_name] = config 

371 

372 def get_health_check_config(self, adapter_name: str) -> dict[str, Any]: 

373 """Get health check configuration for an adapter.""" 

374 return self._check_config.get(adapter_name, {}) 

375 

376 async def schedule_periodic_checks(self, interval_minutes: int = 30) -> None: 

377 """Schedule periodic health checks for all adapters.""" 

378 while True: 

379 try: 

380 await self.check_all_adapters() 

381 await asyncio.sleep(interval_minutes * 60) 

382 except Exception: 

383 # Log error but continue 

384 await asyncio.sleep(60) # Wait 1 minute before retrying