curo1305
c0c0156468
Introduces a standalone plugin system where every integration lives as an independent Python script in ~/.pyra/plugins/, not hardcoded in core. Plugin framework (src/pyra/plugins/): - base.py: Tool dataclass, PyraPlugin Protocol, BasePlugin helper - loader.py: importlib-based discovery; one bad plugin never crashes pyra - registry.py: singleton aggregating tools, slash commands, system prompts - executor.py: approval gate — scans args, prompts y/N, scans result, logs - install.py: copies bundled_plugins/ to ~/.pyra/plugins/ on install Chat integration: - AI tool-use loop (litellm function calling, up to 10 iterations) - Plugin system prompt additions injected per session - Plugin slash commands merged with static commands CLI additions: - pyra plugin list/install/enable/disable/setup - pyra daemon start/stop/status/restart/install/uninstall (stubs for 2.4) Config: PluginConfig + DaemonConfig added to PyraConfig (backwards-compatible) Bootstrap: ~/.pyra/plugins/ and ~/.pyra/logs/ created on startup Security: tool args and results always injection-scanned; plugin dirs validated with assert_safe_path() before loading (symlink protection) Tests: 37 new tests (loader, registry, executor, plugin isolation security) 161 total, all passing. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
152 lines
4.4 KiB
Python
152 lines
4.4 KiB
Python
"""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() == []
|