mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-11 06:48:39 +03:00
295 lines
9.8 KiB
Python
295 lines
9.8 KiB
Python
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nanobot.agent import plugins as agent_plugins
|
|
from nanobot.agent.plugins import (
|
|
AGENT_PLUGIN_MCP_SCHEMA,
|
|
AGENT_PLUGIN_SCHEMA,
|
|
agent_plugin_mcp_servers,
|
|
discover_agent_plugin_states,
|
|
enabled_agent_plugin_skills,
|
|
set_agent_plugin_enabled,
|
|
)
|
|
from nanobot.agent.skills import SkillsLoader
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_plugin_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(
|
|
agent_plugins,
|
|
"get_config_path",
|
|
lambda: tmp_path / "config" / "config.json",
|
|
)
|
|
|
|
|
|
def _write_skill(root: Path, name: str, *, description: str = "Plugin skill.") -> Path:
|
|
skill = root / "skills" / name
|
|
skill.mkdir(parents=True)
|
|
(skill / "SKILL.md").write_text(
|
|
f"---\nname: {name}\ndescription: {description}\n---\n\n# {name}\n",
|
|
encoding="utf-8",
|
|
)
|
|
return skill
|
|
|
|
|
|
def _manifest(name: str, **fields: object) -> dict[str, object]:
|
|
return {"$schema": AGENT_PLUGIN_SCHEMA, "name": name, **fields}
|
|
|
|
|
|
def _write_plugin(
|
|
workspace: Path,
|
|
directory: str,
|
|
*,
|
|
name: str | None = None,
|
|
manifest: dict[str, object] | None = None,
|
|
) -> Path:
|
|
root = workspace / "plugins" / directory
|
|
root.mkdir(parents=True)
|
|
payload = manifest or _manifest(name or directory)
|
|
(root / "plugin.json").write_text(json.dumps(payload), encoding="utf-8")
|
|
return root
|
|
|
|
|
|
def _write_mcp(root: Path, servers: dict[str, object], **fields: object) -> None:
|
|
payload = {"$schema": AGENT_PLUGIN_MCP_SCHEMA, "mcpServers": servers, **fields}
|
|
(root / "mcp.json").write_text(json.dumps(payload), encoding="utf-8")
|
|
|
|
|
|
def _loaded_plugin_skills(workspace: Path) -> list[str]:
|
|
return [name for name, _ in enabled_agent_plugin_skills(workspace)]
|
|
|
|
|
|
def test_skills_loader_discovers_agent_plugin_skill(tmp_path: Path) -> None:
|
|
loader = SkillsLoader(tmp_path, builtin_skills_dir=tmp_path / "builtin")
|
|
assert loader.list_skills() == []
|
|
|
|
plugin = _write_plugin(tmp_path, "acme-tools")
|
|
_write_skill(plugin, "release-notes", description="Draft release notes from changes.")
|
|
set_agent_plugin_enabled(tmp_path, "acme-tools", True)
|
|
|
|
assert loader.list_skills() == [
|
|
{
|
|
"name": "release-notes",
|
|
"path": str(plugin / "skills" / "release-notes" / "SKILL.md"),
|
|
"source": "plugin",
|
|
}
|
|
]
|
|
assert loader.get_explicitly_invoked_skills("Use $release-notes") == ["release-notes"]
|
|
assert "Draft release notes" in (loader.load_skill("release-notes") or "")
|
|
assert "### Agent Plugin skills" in loader.build_skills_summary()
|
|
assert "`acme-tools/skills/release-notes/SKILL.md`" in loader.build_skills_summary()
|
|
|
|
shutil.rmtree(plugin)
|
|
assert loader.list_skills() == []
|
|
assert loader.build_skills_summary() == ""
|
|
|
|
|
|
def test_agent_plugin_skills_are_direct_and_contained(tmp_path: Path) -> None:
|
|
plugin = _write_plugin(tmp_path, "acme-tools")
|
|
_write_skill(plugin, "direct")
|
|
nested = plugin / "skills" / "group" / "nested"
|
|
nested.mkdir(parents=True)
|
|
(nested / "SKILL.md").write_text(
|
|
"---\nname: nested\ndescription: Nested skill.\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
outside = tmp_path / "outside"
|
|
_write_skill(outside, "escaped")
|
|
try:
|
|
(plugin / "skills" / "escaped").symlink_to(
|
|
outside / "skills" / "escaped",
|
|
target_is_directory=True,
|
|
)
|
|
except OSError as exc:
|
|
pytest.skip(f"directory symlink unavailable: {exc}")
|
|
set_agent_plugin_enabled(tmp_path, "acme-tools", True)
|
|
|
|
assert _loaded_plugin_skills(tmp_path) == ["direct"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("manifest", "valid"),
|
|
[
|
|
(
|
|
{"$schema": "https://agent-plugins.org/schemas/2.0.0/plugin.schema.json", "name": "demo"},
|
|
False,
|
|
),
|
|
({"$schema": AGENT_PLUGIN_SCHEMA, "name": "Bad-Name"}, False),
|
|
(
|
|
_manifest(
|
|
"demo",
|
|
futureField=True,
|
|
extensions="invalid but non-fatal",
|
|
),
|
|
True,
|
|
),
|
|
],
|
|
)
|
|
def test_agent_plugin_manifest_failure_boundary(
|
|
tmp_path: Path,
|
|
manifest: dict[str, object],
|
|
valid: bool,
|
|
) -> None:
|
|
plugin = _write_plugin(tmp_path, "demo", manifest=manifest)
|
|
_write_skill(plugin, "example")
|
|
if valid:
|
|
set_agent_plugin_enabled(tmp_path, "demo", True)
|
|
assert bool(discover_agent_plugin_states(tmp_path)) is valid
|
|
assert _loaded_plugin_skills(tmp_path) == (["example"] if valid else [])
|
|
|
|
|
|
def test_agent_plugin_logo_is_validated_and_contained(tmp_path: Path) -> None:
|
|
plugin = _write_plugin(
|
|
tmp_path,
|
|
"demo",
|
|
manifest=_manifest(
|
|
"demo",
|
|
extensions={"dev.nanobot": {"logo": "./assets/icon.png"}},
|
|
),
|
|
)
|
|
assets = plugin / "assets"
|
|
assets.mkdir()
|
|
(assets / "icon.png").write_bytes(b"\x89PNG\r\n\x1a\nlogo")
|
|
outside = tmp_path / "outside.png"
|
|
outside.write_bytes(b"\x89PNG\r\n\x1a\nlogo")
|
|
escaped = _write_plugin(
|
|
tmp_path,
|
|
"escaped",
|
|
manifest=_manifest(
|
|
"escaped",
|
|
extensions={"dev.nanobot": {"logo": "./assets/icon.png"}},
|
|
),
|
|
)
|
|
escaped_assets = escaped / "assets"
|
|
escaped_assets.mkdir()
|
|
try:
|
|
(escaped_assets / "icon.png").symlink_to(outside)
|
|
except OSError as exc:
|
|
pytest.skip(f"file symlink unavailable: {exc}")
|
|
|
|
logos = {state.plugin.name: state.plugin.logo for state in discover_agent_plugin_states(tmp_path)}
|
|
assert logos == {
|
|
"demo": "data:image/png;base64,iVBORw0KGgpsb2dv",
|
|
"escaped": None,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("skill_name", "frontmatter"),
|
|
[
|
|
("wrong-directory", "name: another\ndescription: Mismatch."),
|
|
("missing-description", "name: missing-description"),
|
|
("Bad-Name", "name: Bad-Name\ndescription: Invalid name."),
|
|
],
|
|
)
|
|
def test_invalid_agent_skill_is_skipped(
|
|
tmp_path: Path,
|
|
skill_name: str,
|
|
frontmatter: str,
|
|
) -> None:
|
|
plugin = _write_plugin(tmp_path, "demo")
|
|
skill = plugin / "skills" / skill_name
|
|
skill.mkdir(parents=True)
|
|
(skill / "SKILL.md").write_text(f"---\n{frontmatter}\n---\n", encoding="utf-8")
|
|
set_agent_plugin_enabled(tmp_path, "demo", True)
|
|
|
|
assert _loaded_plugin_skills(tmp_path) == []
|
|
|
|
|
|
def test_skill_precedence_follows_plugin_lifecycle(tmp_path: Path) -> None:
|
|
plugin = _write_plugin(tmp_path, "demo")
|
|
skill = _write_skill(plugin, "shared", description="Plugin version.")
|
|
(skill / "SKILL.md").write_text(
|
|
"---\nname: shared\ndescription: Plugin version.\nalways: true\n---\n\nPlugin body.\n",
|
|
encoding="utf-8",
|
|
)
|
|
builtin = tmp_path / "builtin"
|
|
builtin_skill = builtin / "shared"
|
|
builtin_skill.mkdir(parents=True)
|
|
(builtin_skill / "SKILL.md").write_text(
|
|
"---\nname: shared\ndescription: Built-in version.\n---\n\nBuilt-in body.\n",
|
|
encoding="utf-8",
|
|
)
|
|
workspace_skill = tmp_path / "skills" / "shared"
|
|
workspace_skill.mkdir(parents=True)
|
|
(workspace_skill / "SKILL.md").write_text(
|
|
"---\nname: shared\ndescription: Workspace version.\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
loader = SkillsLoader(tmp_path, builtin_skills_dir=builtin)
|
|
|
|
assert [entry["source"] for entry in loader.list_skills()] == ["workspace"]
|
|
assert "Workspace version" in (loader.load_skill("shared") or "")
|
|
assert loader.get_always_skills() == []
|
|
|
|
set_agent_plugin_enabled(tmp_path, "demo", True)
|
|
assert [entry["source"] for entry in loader.list_skills()] == ["workspace"]
|
|
shutil.rmtree(workspace_skill)
|
|
assert [entry["source"] for entry in loader.list_skills()] == ["plugin"]
|
|
assert "Plugin body" in (loader.load_skill("shared") or "")
|
|
assert loader.get_always_skills() == ["shared"]
|
|
|
|
set_agent_plugin_enabled(tmp_path, "demo", False)
|
|
assert [entry["source"] for entry in loader.list_skills()] == ["builtin"]
|
|
assert "Built-in version" in (loader.load_skill("shared") or "")
|
|
|
|
|
|
def test_plugin_mcp_requires_explicit_enable(tmp_path: Path) -> None:
|
|
plugin = _write_plugin(tmp_path, "desktop")
|
|
executable = plugin / "bin" / "server"
|
|
executable.parent.mkdir()
|
|
executable.write_text("#!/bin/sh\n", encoding="utf-8")
|
|
_write_mcp(
|
|
plugin,
|
|
{
|
|
"desktop": {
|
|
"type": "stdio",
|
|
"command": "./bin/server",
|
|
"args": ["--data", "${PLUGIN_DATA}/state"],
|
|
"cwd": "${PLUGIN_ROOT}",
|
|
},
|
|
"public-http": {"type": "streamable-http", "url": "http://example.com/mcp"},
|
|
"escape": {"type": "stdio", "command": "../outside"},
|
|
},
|
|
)
|
|
|
|
assert agent_plugin_mcp_servers(tmp_path) == {}
|
|
set_agent_plugin_enabled(tmp_path, "desktop", True)
|
|
|
|
servers = agent_plugin_mcp_servers(tmp_path)
|
|
server = servers["desktop"]
|
|
assert server.command == str(executable)
|
|
assert server.cwd == str(plugin)
|
|
assert server.env["PLUGIN_ROOT"] == str(plugin)
|
|
assert server.args[0] == "--data"
|
|
assert server.args[1].endswith("/state")
|
|
|
|
set_agent_plugin_enabled(tmp_path, "desktop", False)
|
|
assert agent_plugin_mcp_servers(tmp_path) == {}
|
|
|
|
|
|
def test_plugin_state_symlink_cannot_escape_config_root(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
config = tmp_path / "config"
|
|
outside = tmp_path / "outside"
|
|
config.mkdir()
|
|
outside.mkdir()
|
|
try:
|
|
(config / "plugin-data").symlink_to(outside, target_is_directory=True)
|
|
except OSError as exc:
|
|
pytest.skip(f"directory symlink unavailable: {exc}")
|
|
monkeypatch.setattr(
|
|
agent_plugins,
|
|
"get_config_path",
|
|
lambda: config / "config.json",
|
|
)
|
|
_write_plugin(tmp_path, "desktop")
|
|
|
|
with pytest.raises(RuntimeError, match="escapes its parent"):
|
|
set_agent_plugin_enabled(tmp_path, "desktop", True)
|