Coverage for integrations / channels / plugins / plugin_system.py: 0.0%

1""" 

2Plugin System for HevolveBot Integration. 

3 

4Provides base Plugin class and PluginManager for loading, unloading, 

5and managing plugins with lifecycle hooks. 

6""" 

7 

8import logging 

9from abc import ABC, abstractmethod 

10from dataclasses import dataclass, field 

11from typing import Any, Callable, Dict, List, Optional 

12from enum import Enum 

13import importlib 

14import sys 

15from datetime import datetime 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20class PluginState(Enum): 

21 """Plugin lifecycle states.""" 

22 UNLOADED = "unloaded" 

23 LOADED = "loaded" 

24 ENABLED = "enabled" 

25 DISABLED = "disabled" 

26 ERROR = "error" 

27 

28 

29@dataclass 

30class PluginMetadata: 

31 """Metadata for a plugin.""" 

32 name: str 

33 version: str 

34 description: str 

35 author: str = "" 

36 dependencies: List[str] = field(default_factory=list) 

37 config_schema: Dict[str, Any] = field(default_factory=dict) 

38 

39 

40class Plugin(ABC): 

41 """ 

42 Base class for all plugins. 

43 

44 Plugins must implement lifecycle hooks and can optionally 

45 implement message processing hooks. 

46 """ 

47 

48 def __init__(self): 

49 self._state = PluginState.UNLOADED 

50 self._config: Dict[str, Any] = {} 

51 self._loaded_at: Optional[datetime] = None 

52 self._error_message: Optional[str] = None 

53 

54 @property 

55 @abstractmethod 

56 def name(self) -> str: 

57 """Return the plugin name.""" 

58 pass 

59 

60 @property 

61 @abstractmethod 

62 def version(self) -> str: 

63 """Return the plugin version.""" 

64 pass 

65 

66 @property 

67 @abstractmethod 

68 def description(self) -> str: 

69 """Return the plugin description.""" 

70 pass 

71 

72 @property 

73 def state(self) -> PluginState: 

74 """Return the current plugin state.""" 

75 return self._state 

76 

77 @property 

78 def config(self) -> Dict[str, Any]: 

79 """Return the plugin configuration.""" 

80 return self._config 

81 

82 @property 

83 def loaded_at(self) -> Optional[datetime]: 

84 """Return when the plugin was loaded.""" 

85 return self._loaded_at 

86 

87 @property 

88 def error_message(self) -> Optional[str]: 

89 """Return any error message if plugin is in error state.""" 

90 return self._error_message 

91 

92 def get_metadata(self) -> PluginMetadata: 

93 """Return plugin metadata.""" 

94 return PluginMetadata( 

95 name=self.name, 

96 version=self.version, 

97 description=self.description 

98 ) 

99 

100 def configure(self, config: Dict[str, Any]) -> None: 

101 """Configure the plugin with given settings.""" 

102 self._config = config 

103 

104 def on_load(self) -> bool: 

105 """ 

106 Called when the plugin is loaded. 

107 

108 Returns: 

109 True if load was successful, False otherwise. 

110 """ 

111 return True 

112 

113 def on_unload(self) -> bool: 

114 """ 

115 Called when the plugin is unloaded. 

116 

117 Returns: 

118 True if unload was successful, False otherwise. 

119 """ 

120 return True 

121 

122 def on_enable(self) -> bool: 

123 """ 

124 Called when the plugin is enabled. 

125 

126 Returns: 

127 True if enable was successful, False otherwise. 

128 """ 

129 return True 

130 

131 def on_disable(self) -> bool: 

132 """ 

133 Called when the plugin is disabled. 

134 

135 Returns: 

136 True if disable was successful, False otherwise. 

137 """ 

138 return True 

139 

140 def on_message(self, message: Dict[str, Any]) -> Optional[Dict[str, Any]]: 

141 """ 

142 Called when a message is received. 

143 

144 Args: 

145 message: The incoming message. 

146 

147 Returns: 

148 Modified message or None to pass through unchanged. 

149 """ 

150 return None 

151 

152 def on_response(self, response: Dict[str, Any]) -> Optional[Dict[str, Any]]: 

153 """ 

154 Called when a response is about to be sent. 

155 

156 Args: 

157 response: The outgoing response. 

158 

159 Returns: 

160 Modified response or None to pass through unchanged. 

161 """ 

162 return None 

163 

164 def on_error(self, error: Exception) -> None: 

165 """ 

166 Called when an error occurs during message processing. 

167 

168 Args: 

169 error: The exception that occurred. 

170 """ 

171 pass 

172 

173 

174class PluginManager: 

175 """ 

176 Manages plugin lifecycle and message routing. 

177 

178 Handles loading, unloading, enabling, and disabling plugins, 

179 as well as routing messages through active plugins. 

180 """ 

181 

182 def __init__(self): 

183 self._plugins: Dict[str, Plugin] = {} 

184 self._load_order: List[str] = [] 

185 self._message_hooks: List[Callable] = [] 

186 self._response_hooks: List[Callable] = [] 

187 

188 @property 

189 def plugins(self) -> Dict[str, Plugin]: 

190 """Return all loaded plugins.""" 

191 return self._plugins.copy() 

192 

193 def load(self, plugin: Plugin, config: Optional[Dict[str, Any]] = None) -> bool: 

194 """ 

195 Load a plugin. 

196 

197 Args: 

198 plugin: The plugin instance to load. 

199 config: Optional configuration for the plugin. 

200 

201 Returns: 

202 True if load was successful, False otherwise. 

203 """ 

204 if plugin.name in self._plugins: 

205 logger.warning(f"Plugin {plugin.name} is already loaded") 

206 return False 

207 

208 try: 

209 if config: 

210 plugin.configure(config) 

211 

212 if plugin.on_load(): 

213 plugin._state = PluginState.LOADED 

214 plugin._loaded_at = datetime.utcnow() 

215 self._plugins[plugin.name] = plugin 

216 self._load_order.append(plugin.name) 

217 logger.info(f"Plugin {plugin.name} v{plugin.version} loaded successfully") 

218 return True 

219 else: 

220 plugin._state = PluginState.ERROR 

221 plugin._error_message = "on_load() returned False" 

222 logger.error(f"Plugin {plugin.name} failed to load") 

223 return False 

224 except Exception as e: 

225 plugin._state = PluginState.ERROR 

226 plugin._error_message = str(e) 

227 logger.exception(f"Error loading plugin {plugin.name}: {e}") 

228 return False 

229 

230 def load_from_module(self, module_path: str, class_name: str, 

231 config: Optional[Dict[str, Any]] = None) -> bool: 

232 """ 

233 Load a plugin from a module path. 

234 

235 Args: 

236 module_path: The module path (e.g., 'my.plugins.example'). 

237 class_name: The plugin class name. 

238 config: Optional configuration for the plugin. 

239 

240 Returns: 

241 True if load was successful, False otherwise. 

242 """ 

243 try: 

244 module = importlib.import_module(module_path) 

245 plugin_class = getattr(module, class_name) 

246 plugin = plugin_class() 

247 return self.load(plugin, config) 

248 except Exception as e: 

249 logger.exception(f"Error loading plugin from {module_path}.{class_name}: {e}") 

250 return False 

251 

252 def unload(self, plugin_name: str) -> bool: 

253 """ 

254 Unload a plugin. 

255 

256 Args: 

257 plugin_name: The name of the plugin to unload. 

258 

259 Returns: 

260 True if unload was successful, False otherwise. 

261 """ 

262 if plugin_name not in self._plugins: 

263 logger.warning(f"Plugin {plugin_name} is not loaded") 

264 return False 

265 

266 plugin = self._plugins[plugin_name] 

267 

268 try: 

269 # Disable first if enabled 

270 if plugin.state == PluginState.ENABLED: 

271 self.disable(plugin_name) 

272 

273 if plugin.on_unload(): 

274 plugin._state = PluginState.UNLOADED 

275 del self._plugins[plugin_name] 

276 self._load_order.remove(plugin_name) 

277 logger.info(f"Plugin {plugin_name} unloaded successfully") 

278 return True 

279 else: 

280 logger.error(f"Plugin {plugin_name} failed to unload") 

281 return False 

282 except Exception as e: 

283 logger.exception(f"Error unloading plugin {plugin_name}: {e}") 

284 return False 

285 

286 def enable(self, plugin_name: str) -> bool: 

287 """ 

288 Enable a loaded plugin. 

289 

290 Args: 

291 plugin_name: The name of the plugin to enable. 

292 

293 Returns: 

294 True if enable was successful, False otherwise. 

295 """ 

296 if plugin_name not in self._plugins: 

297 logger.warning(f"Plugin {plugin_name} is not loaded") 

298 return False 

299 

300 plugin = self._plugins[plugin_name] 

301 

302 if plugin.state == PluginState.ENABLED: 

303 logger.warning(f"Plugin {plugin_name} is already enabled") 

304 return True 

305 

306 if plugin.state not in (PluginState.LOADED, PluginState.DISABLED): 

307 logger.error(f"Plugin {plugin_name} is in invalid state: {plugin.state}") 

308 return False 

309 

310 try: 

311 if plugin.on_enable(): 

312 plugin._state = PluginState.ENABLED 

313 # Register hooks 

314 if hasattr(plugin, 'on_message'): 

315 self._message_hooks.append(plugin.on_message) 

316 if hasattr(plugin, 'on_response'): 

317 self._response_hooks.append(plugin.on_response) 

318 logger.info(f"Plugin {plugin_name} enabled") 

319 return True 

320 else: 

321 logger.error(f"Plugin {plugin_name} failed to enable") 

322 return False 

323 except Exception as e: 

324 logger.exception(f"Error enabling plugin {plugin_name}: {e}") 

325 return False 

326 

327 def disable(self, plugin_name: str) -> bool: 

328 """ 

329 Disable an enabled plugin. 

330 

331 Args: 

332 plugin_name: The name of the plugin to disable. 

333 

334 Returns: 

335 True if disable was successful, False otherwise. 

336 """ 

337 if plugin_name not in self._plugins: 

338 logger.warning(f"Plugin {plugin_name} is not loaded") 

339 return False 

340 

341 plugin = self._plugins[plugin_name] 

342 

343 if plugin.state != PluginState.ENABLED: 

344 logger.warning(f"Plugin {plugin_name} is not enabled") 

345 return True 

346 

347 try: 

348 if plugin.on_disable(): 

349 plugin._state = PluginState.DISABLED 

350 # Unregister hooks 

351 if plugin.on_message in self._message_hooks: 

352 self._message_hooks.remove(plugin.on_message) 

353 if plugin.on_response in self._response_hooks: 

354 self._response_hooks.remove(plugin.on_response) 

355 logger.info(f"Plugin {plugin_name} disabled") 

356 return True 

357 else: 

358 logger.error(f"Plugin {plugin_name} failed to disable") 

359 return False 

360 except Exception as e: 

361 logger.exception(f"Error disabling plugin {plugin_name}: {e}") 

362 return False 

363 

364 def list_plugins(self) -> List[Dict[str, Any]]: 

365 """ 

366 List all loaded plugins with their status. 

367 

368 Returns: 

369 List of plugin information dictionaries. 

370 """ 

371 result = [] 

372 for name in self._load_order: 

373 plugin = self._plugins[name] 

374 result.append({ 

375 "name": plugin.name, 

376 "version": plugin.version, 

377 "description": plugin.description, 

378 "state": plugin.state.value, 

379 "loaded_at": plugin.loaded_at.isoformat() if plugin.loaded_at else None, 

380 "error": plugin.error_message 

381 }) 

382 return result 

383 

384 def get_plugin(self, plugin_name: str) -> Optional[Plugin]: 

385 """ 

386 Get a plugin by name. 

387 

388 Args: 

389 plugin_name: The name of the plugin. 

390 

391 Returns: 

392 The plugin instance or None if not found. 

393 """ 

394 return self._plugins.get(plugin_name) 

395 

396 def process_message(self, message: Dict[str, Any]) -> Dict[str, Any]: 

397 """ 

398 Process a message through all enabled plugins. 

399 

400 Args: 

401 message: The incoming message. 

402 

403 Returns: 

404 The processed message. 

405 """ 

406 current_message = message.copy() 

407 

408 for hook in self._message_hooks: 

409 try: 

410 result = hook(current_message) 

411 if result is not None: 

412 current_message = result 

413 except Exception as e: 

414 logger.exception(f"Error in message hook: {e}") 

415 

416 return current_message 

417 

418 def process_response(self, response: Dict[str, Any]) -> Dict[str, Any]: 

419 """ 

420 Process a response through all enabled plugins. 

421 

422 Args: 

423 response: The outgoing response. 

424 

425 Returns: 

426 The processed response. 

427 """ 

428 current_response = response.copy() 

429 

430 for hook in self._response_hooks: 

431 try: 

432 result = hook(current_response) 

433 if result is not None: 

434 current_response = result 

435 except Exception as e: 

436 logger.exception(f"Error in response hook: {e}") 

437 

438 return current_response 

439 

440 def reload(self, plugin_name: str) -> bool: 

441 """ 

442 Reload a plugin. 

443 

444 Args: 

445 plugin_name: The name of the plugin to reload. 

446 

447 Returns: 

448 True if reload was successful, False otherwise. 

449 """ 

450 if plugin_name not in self._plugins: 

451 logger.warning(f"Plugin {plugin_name} is not loaded") 

452 return False 

453 

454 plugin = self._plugins[plugin_name] 

455 was_enabled = plugin.state == PluginState.ENABLED 

456 config = plugin.config.copy() 

457 

458 # Store plugin class for reinstantiation 

459 plugin_class = plugin.__class__ 

460 

461 if not self.unload(plugin_name): 

462 return False 

463 

464 # Create new instance 

465 new_plugin = plugin_class() 

466 

467 if not self.load(new_plugin, config): 

468 return False 

469 

470 if was_enabled: 

471 return self.enable(plugin_name) 

472 

473 return True 

474 

475 def unload_all(self) -> bool: 

476 """ 

477 Unload all plugins. 

478 

479 Returns: 

480 True if all plugins were unloaded successfully. 

481 """ 

482 success = True 

483 # Unload in reverse order 

484 for name in reversed(self._load_order.copy()): 

485 if not self.unload(name): 

486 success = False 

487 return success