Coverage for merco/plugins/manager.py: 79%

1"""PluginManager - plugin lifecycle management""" 

2from __future__ import annotations 

3 

4import logging 

5from typing import TYPE_CHECKING 

6 

7if TYPE_CHECKING: 

8 from .base import Plugin, PluginContext 

9 

10logger = logging.getLogger("merco.plugins.manager") 

11 

12 

13class PluginManager: 

14 """Plugin discovery, loading, activation, and deactivation""" 

15 

16 def __init__(self, ctx: "PluginContext"): 

17 self._ctx = ctx 

18 self._plugins: dict[str, "Plugin"] = {} 

19 self._active: set[str] = set() 

20 

21 def register(self, plugin: "Plugin") -> None: 

22 """Register a plugin instance""" 

23 self._plugins[plugin.name] = plugin 

24 

25 async def activate(self, name: str) -> None: 

26 """Activate a single plugin""" 

27 if name in self._active: 

28 return 

29 plugin = self._plugins.get(name) 

30 if not plugin: 

31 logger.warning("Plugin '%s' not registered", name) 

32 return 

33 try: 

34 await plugin.activate(self._ctx) 

35 self._active.add(name) 

36 await self._ctx.hooks.emit("plugin.activated", plugin_name=name, version=plugin.version) 

37 except Exception as e: 

38 logger.warning("Plugin '%s' activation failed: %s", name, e) 

39 try: 

40 await self._ctx.hooks.emit("plugin.error", plugin_name=name, error=str(e)) 

41 except Exception: 

42 pass 

43 

44 async def deactivate(self, name: str) -> None: 

45 """Deactivate a single plugin""" 

46 plugin = self._plugins.get(name) 

47 if not plugin: 

48 return 

49 try: 

50 await plugin.deactivate() 

51 except Exception as e: 

52 logger.warning("Plugin '%s' deactivation failed: %s", name, e) 

53 self._active.discard(name) 

54 try: 

55 await self._ctx.hooks.emit("plugin.deactivated", plugin_name=name) 

56 except Exception: 

57 pass 

58 

59 async def activate_all(self) -> None: 

60 """Activate all enabled plugins at startup""" 

61 plugins_config = getattr(self._ctx.config, "plugins", {}) 

62 for name in self._plugins: 

63 plugin_cfg = plugins_config.get(name, {}) 

64 if plugin_cfg.get("enabled", True): 

65 await self.activate(name) 

66 

67 async def deactivate_all(self) -> None: 

68 """Deactivate all active plugins""" 

69 for name in list(self._active): 

70 await self.deactivate(name) 

71 

72 @property 

73 def active_plugins(self) -> list[str]: 

74 """Return list of active plugin names""" 

75 return list(self._active)