"""Tests for PluginRegistry aggregation and singleton behavior.""" import json from pathlib import Path import pytest from pyra.plugins.base import BasePlugin, Tool from pyra.plugins.registry import PluginRegistry def _make_plugin_dir(plugins_dir: Path, name: str, plugin_code: str) -> None: d = plugins_dir / name d.mkdir(parents=True) (d / "manifest.json").write_text(json.dumps({"name": name, "version": "1.0.0"})) (d / "plugin.py").write_text(plugin_code) _ALPHA_PLUGIN = """\ from pyra.plugins.base import BasePlugin, Tool class _P(BasePlugin): name = "alpha" description = "Alpha plugin" version = "1.0.0" def tools(self): return [ Tool("alpha_tool", "An alpha tool", {"type": "object", "properties": {}}, lambda: "alpha result", requires_approval=False) ] def slash_commands(self): return {"/alpha": lambda: None} def system_prompt_addition(self): return "Alpha is active." def get_plugin(): return _P() """ _BETA_PLUGIN = """\ from pyra.plugins.base import BasePlugin, Tool class _P(BasePlugin): name = "beta" description = "Beta plugin" version = "1.0.0" def tools(self): return [ Tool("beta_tool", "A beta tool", {"type": "object", "properties": {}}, lambda: "beta result", requires_approval=True) ] def system_prompt_addition(self): return "Beta is active." def get_plugin(): return _P() """ def test_singleton_returns_same_instance(tmp_pyra_home): r1 = PluginRegistry.instance() r2 = PluginRegistry.instance() assert r1 is r2 def test_load_all_only_loads_enabled(tmp_pyra_home, tmp_path): plugins_dir = tmp_path / "plugins" plugins_dir.mkdir() _make_plugin_dir(plugins_dir, "alpha", _ALPHA_PLUGIN) _make_plugin_dir(plugins_dir, "beta", _BETA_PLUGIN) registry = PluginRegistry.instance() registry.load_all(plugins_dir, enabled_names=["alpha"]) names = {p.name for p in registry.get_active_plugins()} assert "alpha" in names assert "beta" not in names def test_get_all_tools_aggregates(tmp_pyra_home, tmp_path): plugins_dir = tmp_path / "plugins" plugins_dir.mkdir() _make_plugin_dir(plugins_dir, "alpha", _ALPHA_PLUGIN) _make_plugin_dir(plugins_dir, "beta", _BETA_PLUGIN) registry = PluginRegistry.instance() registry.load_all(plugins_dir, enabled_names=["alpha", "beta"]) tool_names = {t.name for t in registry.get_all_tools()} assert "alpha_tool" in tool_names assert "beta_tool" in tool_names def test_get_slash_commands_aggregates(tmp_pyra_home, tmp_path): plugins_dir = tmp_path / "plugins" plugins_dir.mkdir() _make_plugin_dir(plugins_dir, "alpha", _ALPHA_PLUGIN) registry = PluginRegistry.instance() registry.load_all(plugins_dir, enabled_names=["alpha"]) cmds = registry.get_slash_commands() assert "/alpha" in cmds def test_get_system_prompt_additions(tmp_pyra_home, tmp_path): plugins_dir = tmp_path / "plugins" plugins_dir.mkdir() _make_plugin_dir(plugins_dir, "alpha", _ALPHA_PLUGIN) _make_plugin_dir(plugins_dir, "beta", _BETA_PLUGIN) registry = PluginRegistry.instance() registry.load_all(plugins_dir, enabled_names=["alpha", "beta"]) additions = registry.get_system_prompt_additions() assert "Alpha is active." in additions assert "Beta is active." in additions def test_find_tool_returns_correct_tool(tmp_pyra_home, tmp_path): plugins_dir = tmp_path / "plugins" plugins_dir.mkdir() _make_plugin_dir(plugins_dir, "alpha", _ALPHA_PLUGIN) registry = PluginRegistry.instance() registry.load_all(plugins_dir, enabled_names=["alpha"]) tool = registry.find_tool("alpha_tool") assert tool is not None assert tool.name == "alpha_tool" def test_find_tool_unknown_returns_none(tmp_pyra_home): registry = PluginRegistry.instance() registry.load_all(Path("/nonexistent"), enabled_names=[]) assert registry.find_tool("no_such_tool") is None def test_empty_registry_returns_empty_collections(tmp_pyra_home): registry = PluginRegistry.instance() registry.load_all(Path("/nonexistent"), enabled_names=[]) assert registry.get_all_tools() == [] assert registry.get_slash_commands() == {} assert registry.get_system_prompt_additions() == "" assert registry.get_active_plugins() == []