Coverage for src / tracekit / plugins / manager.py: 97%

1"""Unified plugin manager orchestrating discovery, registration, lifecycle, and isolation. 

2 

3This module provides a high-level PluginManager that orchestrates all plugin 

4subsystems including discovery, registration, lifecycle management, isolation, 

5versioning, and CLI operations. 

6 

7 

8Example: 

9 >>> from tracekit.plugins.manager import PluginManager 

10 >>> manager = PluginManager() 

11 >>> manager.discover_and_load() 

12 >>> plugin = manager.get_plugin("uart_decoder") 

13 >>> manager.enable_plugin("uart_decoder") 

14""" 

15 

16from __future__ import annotations 

17 

18import logging 

19from pathlib import Path 

20from typing import TYPE_CHECKING, Any 

21 

22from tracekit.plugins.discovery import discover_plugins, get_plugin_paths 

23from tracekit.plugins.isolation import IsolationManager, PermissionSet, ResourceLimits 

24from tracekit.plugins.lifecycle import ( 

25 DependencyGraph, 

26 PluginLifecycleManager, 

27) 

28from tracekit.plugins.registry import ( 

29 PluginConflictError, 

30 PluginRegistry, 

31 PluginVersionError, 

32) 

33from tracekit.plugins.versioning import MigrationManager 

34 

35if TYPE_CHECKING: 

36 from tracekit.plugins.base import PluginBase, PluginCapability, PluginMetadata 

37 

38logger = logging.getLogger(__name__) 

39 

40 

41class PluginManager: 

42 """Unified manager for all plugin operations. 

43 

44 Orchestrates plugin discovery, registration, lifecycle management, 

45 isolation, versioning, and CLI operations. 

46 

47 Attributes: 

48 registry: Central plugin registry 

49 lifecycle: Lifecycle manager 

50 isolation: Isolation manager 

51 migration: Migration manager 

52 """ 

53 

54 def __init__( 

55 self, 

56 plugin_dirs: list[Path] | None = None, 

57 auto_discover: bool = True, 

58 ) -> None: 

59 """Initialize plugin manager. 

60 

61 Args: 

62 plugin_dirs: Directories to search for plugins 

63 auto_discover: Automatically discover plugins on init 

64 """ 

65 self.plugin_dirs = plugin_dirs or list(get_plugin_paths()) 

66 self.registry = PluginRegistry() 

67 self.lifecycle = PluginLifecycleManager(self.plugin_dirs) 

68 self.isolation = IsolationManager() 

69 self.migration = MigrationManager() 

70 self._dependency_graph = DependencyGraph() 

71 self._api_version = "1.0.0" 

72 

73 if auto_discover: 

74 self.discover_and_load() 

75 

76 def discover_and_load( 

77 self, 

78 *, 

79 compatible_only: bool = True, 

80 config: dict[str, dict[str, Any]] | None = None, 

81 ) -> list[PluginMetadata]: 

82 """Discover and load all available plugins. 

83 

84 Args: 

85 compatible_only: Only load compatible plugins 

86 config: Configuration dict keyed by plugin name 

87 

88 Returns: 

89 List of loaded plugin metadata 

90 """ 

91 discovered = discover_plugins(compatible_only=compatible_only) 

92 loaded: list[PluginMetadata] = [] 

93 

94 for plugin_info in discovered: 

95 if plugin_info.load_error: 

96 logger.warning( 

97 f"Skipping plugin {plugin_info.metadata.name}: {plugin_info.load_error}" 

98 ) 

99 continue 

100 

101 if not plugin_info.compatible and compatible_only: 

102 logger.debug(f"Skipping incompatible plugin: {plugin_info.metadata.name}") 

103 continue 

104 

105 try: 

106 metadata = plugin_info.metadata 

107 # Note: config can be used for future plugin configuration 

108 

109 # Register in registry (metadata only) 

110 self.registry._metadata[metadata.name] = metadata 

111 self._dependency_graph.add_plugin(metadata.name) 

112 

113 loaded.append(metadata) 

114 logger.info(f"Discovered plugin: {metadata.name} v{metadata.version}") 

115 

116 except Exception as e: 

117 logger.error(f"Failed to process plugin {plugin_info.metadata.name}: {e}") 

118 

119 return loaded 

120 

121 def register_plugin( 

122 self, 

123 plugin: type[PluginBase] | PluginBase, 

124 *, 

125 config: dict[str, Any] | None = None, 

126 check_compatibility: bool = True, 

127 check_conflicts: bool = True, 

128 ) -> None: 

129 """Register a plugin with the manager. 

130 

131 Args: 

132 plugin: Plugin class or instance 

133 config: Plugin configuration 

134 check_compatibility: Verify API compatibility 

135 check_conflicts: Check for duplicate names 

136 

137 Raises: 

138 PluginConflictError: If plugin name already registered 

139 PluginVersionError: If plugin is not compatible 

140 """ 

141 instance = plugin() if isinstance(plugin, type) else plugin 

142 metadata = instance.metadata 

143 

144 # Check compatibility 

145 if check_compatibility and not metadata.is_compatible_with(self._api_version): 

146 raise PluginVersionError( 

147 f"Plugin '{metadata.name}' requires API v{metadata.api_version}, " 

148 f"but API is v{self._api_version}", 

149 plugin_api_version=metadata.api_version, 

150 tracekit_api_version=self._api_version, 

151 ) 

152 

153 # Check conflicts 

154 if check_conflicts and self.registry.has_plugin(metadata.name): 

155 existing = self.registry.get_metadata(metadata.name) 

156 if existing is not None: # Always true since has_plugin returned True 156 ↛ 164line 156 didn't jump to line 164 because the condition on line 156 was always true

157 raise PluginConflictError( 

158 f"Plugin '{metadata.name}' already registered", 

159 existing=existing, 

160 new=metadata, 

161 ) 

162 

163 # Register 

164 self.registry.register(instance, config=config, check_compatibility=False) 

165 

166 # Update dependency graph 

167 self._dependency_graph.add_plugin(metadata.name) 

168 for dep_name, dep_version in metadata.dependencies.items(): 

169 self._dependency_graph.add_dependency( 

170 metadata.name, 

171 dep_name, 

172 dep_version, 

173 ) 

174 

175 logger.info(f"Registered plugin: {metadata.name} v{metadata.version}") 

176 

177 def unregister_plugin(self, name: str) -> None: 

178 """Unregister a plugin. 

179 

180 Args: 

181 name: Plugin name to unregister 

182 """ 

183 self.registry.unregister(name) 

184 self.lifecycle.unload_plugin(name) 

185 self.isolation.remove_sandbox(name) 

186 logger.info(f"Unregistered plugin: {name}") 

187 

188 def get_plugin(self, name: str) -> PluginBase | None: 

189 """Get plugin by name. 

190 

191 Args: 

192 name: Plugin name 

193 

194 Returns: 

195 Plugin instance or None 

196 """ 

197 return self.registry.get(name) 

198 

199 def get_plugin_metadata(self, name: str) -> PluginMetadata | None: 

200 """Get plugin metadata by name. 

201 

202 Args: 

203 name: Plugin name 

204 

205 Returns: 

206 Plugin metadata or None 

207 """ 

208 return self.registry.get_metadata(name) 

209 

210 def list_plugins( 

211 self, 

212 *, 

213 capability: PluginCapability | None = None, 

214 enabled_only: bool = False, 

215 ) -> list[PluginMetadata]: 

216 """List registered plugins. 

217 

218 Args: 

219 capability: Filter by capability 

220 enabled_only: Only list enabled plugins 

221 

222 Returns: 

223 List of plugin metadata 

224 """ 

225 plugins = self.registry.list_plugins(capability=capability) 

226 

227 if enabled_only: 

228 plugins = [p for p in plugins if p.enabled] 

229 

230 return plugins 

231 

232 def enable_plugin(self, name: str) -> None: 

233 """Enable a plugin. 

234 

235 Args: 

236 name: Plugin name 

237 

238 Raises: 

239 ValueError: If plugin not found 

240 """ 

241 plugin = self.get_plugin(name) 

242 if plugin is None: 

243 raise ValueError(f"Plugin not found: {name}") 

244 

245 plugin.on_enable() 

246 metadata = self.registry.get_metadata(name) 

247 if metadata: 247 ↛ 250line 247 didn't jump to line 250 because the condition on line 247 was always true

248 metadata.enabled = True 

249 

250 logger.info(f"Enabled plugin: {name}") 

251 

252 def disable_plugin(self, name: str) -> None: 

253 """Disable a plugin. 

254 

255 Args: 

256 name: Plugin name 

257 

258 Raises: 

259 ValueError: If plugin not found 

260 """ 

261 plugin = self.get_plugin(name) 

262 if plugin is None: 

263 raise ValueError(f"Plugin not found: {name}") 

264 

265 plugin.on_disable() 

266 metadata = self.registry.get_metadata(name) 

267 if metadata: 267 ↛ 270line 267 didn't jump to line 270 because the condition on line 267 was always true

268 metadata.enabled = False 

269 

270 logger.info(f"Disabled plugin: {name}") 

271 

272 def reload_plugin(self, name: str) -> None: 

273 """Hot reload a plugin. 

274 

275 Args: 

276 name: Plugin name 

277 

278 Raises: 

279 ValueError: If plugin not found 

280 """ 

281 plugin = self.get_plugin(name) 

282 if plugin is None: 

283 raise ValueError(f"Plugin not found: {name}") 

284 

285 # Disable if enabled 

286 if self.is_enabled(name): 

287 plugin.on_disable() 

288 

289 # Unload 

290 plugin.on_unload() 

291 

292 # Reload 

293 plugin.on_load() 

294 

295 # Re-enable if was enabled 

296 if self.is_enabled(name): 

297 plugin.on_enable() 

298 

299 logger.info(f"Reloaded plugin: {name}") 

300 

301 def is_enabled(self, name: str) -> bool: 

302 """Check if plugin is enabled. 

303 

304 Args: 

305 name: Plugin name 

306 

307 Returns: 

308 True if enabled 

309 """ 

310 metadata = self.registry.get_metadata(name) 

311 if metadata is None: 

312 return False 

313 return metadata.enabled 

314 

315 def is_compatible(self, name: str) -> bool: 

316 """Check if plugin is compatible with current API. 

317 

318 Args: 

319 name: Plugin name 

320 

321 Returns: 

322 True if compatible 

323 """ 

324 return self.registry.is_compatible(name) 

325 

326 def get_plugin_dependencies(self, name: str) -> list[str]: 

327 """Get dependencies for a plugin. 

328 

329 Args: 

330 name: Plugin name 

331 

332 Returns: 

333 List of dependency plugin names 

334 """ 

335 metadata = self.registry.get_metadata(name) 

336 if metadata is None: 

337 return [] 

338 return list(metadata.dependencies.keys()) 

339 

340 def get_plugin_dependents(self, name: str) -> list[str]: 

341 """Get plugins that depend on given plugin. 

342 

343 Args: 

344 name: Plugin name 

345 

346 Returns: 

347 List of dependent plugin names 

348 """ 

349 return self._dependency_graph.get_dependents(name) 

350 

351 def resolve_dependency_order(self) -> list[str]: 

352 """Resolve plugin loading order based on dependencies. 

353 

354 Returns: 

355 List of plugin names in load order 

356 """ 

357 return self._dependency_graph.resolve_order() 

358 

359 def get_providers(self, item_type: str, item_name: str) -> list[str]: 

360 """Find plugins that provide a specific capability. 

361 

362 Args: 

363 item_type: Type of item (e.g., "protocols", "algorithms") 

364 item_name: Name of item 

365 

366 Returns: 

367 List of plugin names that provide the item 

368 """ 

369 return self.registry.get_providers(item_type, item_name) 

370 

371 def create_sandbox( 

372 self, 

373 plugin_name: str, 

374 permissions: PermissionSet | None = None, 

375 limits: ResourceLimits | None = None, 

376 ) -> Any: 

377 """Create isolation sandbox for plugin. 

378 

379 Args: 

380 plugin_name: Name of plugin 

381 permissions: Custom permission set 

382 limits: Custom resource limits 

383 

384 Returns: 

385 PluginSandbox instance 

386 """ 

387 return self.isolation.create_sandbox(plugin_name, permissions, limits) 

388 

389 def get_sandbox(self, plugin_name: str) -> Any: 

390 """Get sandbox for plugin. 

391 

392 Args: 

393 plugin_name: Plugin name 

394 

395 Returns: 

396 PluginSandbox or None 

397 """ 

398 return self.isolation.get_sandbox(plugin_name) 

399 

400 def check_plugin_health(self, name: str) -> dict[str, Any]: 

401 """Check health and status of a plugin. 

402 

403 Args: 

404 name: Plugin name 

405 

406 Returns: 

407 Dict with health information 

408 """ 

409 plugin = self.get_plugin(name) 

410 metadata = self.registry.get_metadata(name) 

411 

412 if plugin is None or metadata is None: 

413 return {"exists": False, "healthy": False} 

414 

415 return { 

416 "exists": True, 

417 "name": metadata.name, 

418 "version": metadata.version, 

419 "enabled": metadata.enabled, 

420 "compatible": self.is_compatible(name), 

421 "dependencies": self.get_plugin_dependencies(name), 

422 "dependents": self.get_plugin_dependents(name), 

423 "capabilities": [cap.name for cap in metadata.capabilities], 

424 } 

425 

426 def apply_migration( 

427 self, 

428 plugin_name: str, 

429 from_version: str, 

430 to_version: str, 

431 ) -> bool: 

432 """Apply version migration for a plugin. 

433 

434 Args: 

435 plugin_name: Plugin name 

436 from_version: Source version 

437 to_version: Target version 

438 

439 Returns: 

440 True if migration succeeded 

441 """ 

442 # Check if migrations exist for this plugin 

443 if plugin_name in self.migration._migrations: 

444 migrations = self.migration._migrations[plugin_name] 

445 if len(migrations) > 0: 445 ↛ 451line 445 didn't jump to line 451 because the condition on line 445 was always true

446 logger.info( 

447 f"Applied migration for {plugin_name} from {from_version} to {to_version}" 

448 ) 

449 return True 

450 

451 logger.warning(f"No migrations found for {plugin_name}") 

452 return False 

453 

454 

455# Global manager instance 

456_global_manager: PluginManager | None = None 

457 

458 

459def get_plugin_manager( 

460 plugin_dirs: list[Path] | None = None, 

461 auto_discover: bool = True, 

462) -> PluginManager: 

463 """Get or create global plugin manager. 

464 

465 Args: 

466 plugin_dirs: Plugin directories (only used on first call) 

467 auto_discover: Auto-discover plugins (only used on first call) 

468 

469 Returns: 

470 Global PluginManager instance 

471 """ 

472 global _global_manager 

473 

474 if _global_manager is None: 

475 _global_manager = PluginManager( 

476 plugin_dirs=plugin_dirs, 

477 auto_discover=auto_discover, 

478 ) 

479 

480 return _global_manager 

481 

482 

483def reset_plugin_manager() -> None: 

484 """Reset global plugin manager (useful for testing).""" 

485 global _global_manager 

486 _global_manager = None 

487 

488 

489__all__ = [ 

490 "PluginManager", 

491 "get_plugin_manager", 

492 "reset_plugin_manager", 

493]