refactor(agent): streamline hook method calls and enhance error logging
- Introduced a helper method `_for_each_hook_safe` to reduce code duplication in hook method implementations. - Updated error logging to include the method name for better traceability. - Improved the `SkillsLoader` class by adding a new method `_skill_entries_from_dir` to simplify skill listing logic. - Enhanced skill loading and filtering logic, ensuring workspace skills take precedence over built-in ones. - Added comprehensive tests for `SkillsLoader` to validate functionality and edge cases.
This commit is contained in:
Jack Lu
Xubin Ren
@@ -0,0 +1,252 @@
|
||||
"""Tests for nanobot.agent.skills.SkillsLoader."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from nanobot.agent.skills import SkillsLoader
|
||||
|
||||
|
||||
def _write_skill(
|
||||
base: Path,
|
||||
name: str,
|
||||
*,
|
||||
metadata_json: dict | None = None,
|
||||
body: str = "# Skill\n",
|
||||
) -> Path:
|
||||
"""Create ``base / name / SKILL.md`` with optional nanobot metadata JSON."""
|
||||
skill_dir = base / name
|
||||
skill_dir.mkdir(parents=True)
|
||||
lines = ["---"]
|
||||
if metadata_json is not None:
|
||||
payload = json.dumps({"nanobot": metadata_json}, separators=(",", ":"))
|
||||
lines.append(f'metadata: {payload}')
|
||||
lines.extend(["---", "", body])
|
||||
path = skill_dir / "SKILL.md"
|
||||
path.write_text("\n".join(lines), encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def test_list_skills_empty_when_skills_dir_missing(tmp_path: Path) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
workspace.mkdir()
|
||||
builtin = tmp_path / "builtin"
|
||||
builtin.mkdir()
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
assert loader.list_skills(filter_unavailable=False) == []
|
||||
|
||||
|
||||
def test_list_skills_empty_when_skills_dir_exists_but_empty(tmp_path: Path) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
(workspace / "skills").mkdir(parents=True)
|
||||
builtin = tmp_path / "builtin"
|
||||
builtin.mkdir()
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
assert loader.list_skills(filter_unavailable=False) == []
|
||||
|
||||
|
||||
def test_list_skills_workspace_entry_shape_and_source(tmp_path: Path) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
skills_root = workspace / "skills"
|
||||
skills_root.mkdir(parents=True)
|
||||
skill_path = _write_skill(skills_root, "alpha", body="# Alpha")
|
||||
builtin = tmp_path / "builtin"
|
||||
builtin.mkdir()
|
||||
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
entries = loader.list_skills(filter_unavailable=False)
|
||||
assert entries == [
|
||||
{"name": "alpha", "path": str(skill_path), "source": "workspace"},
|
||||
]
|
||||
|
||||
|
||||
def test_list_skills_skips_non_directories_and_missing_skill_md(tmp_path: Path) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
skills_root = workspace / "skills"
|
||||
skills_root.mkdir(parents=True)
|
||||
(skills_root / "not_a_dir.txt").write_text("x", encoding="utf-8")
|
||||
(skills_root / "no_skill_md").mkdir()
|
||||
ok_path = _write_skill(skills_root, "ok", body="# Ok")
|
||||
builtin = tmp_path / "builtin"
|
||||
builtin.mkdir()
|
||||
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
entries = loader.list_skills(filter_unavailable=False)
|
||||
names = {entry["name"] for entry in entries}
|
||||
assert names == {"ok"}
|
||||
assert entries[0]["path"] == str(ok_path)
|
||||
|
||||
|
||||
def test_list_skills_workspace_shadows_builtin_same_name(tmp_path: Path) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
ws_skills = workspace / "skills"
|
||||
ws_skills.mkdir(parents=True)
|
||||
ws_path = _write_skill(ws_skills, "dup", body="# Workspace wins")
|
||||
|
||||
builtin = tmp_path / "builtin"
|
||||
_write_skill(builtin, "dup", body="# Builtin")
|
||||
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
entries = loader.list_skills(filter_unavailable=False)
|
||||
assert len(entries) == 1
|
||||
assert entries[0]["source"] == "workspace"
|
||||
assert entries[0]["path"] == str(ws_path)
|
||||
|
||||
|
||||
def test_list_skills_merges_workspace_and_builtin(tmp_path: Path) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
ws_skills = workspace / "skills"
|
||||
ws_skills.mkdir(parents=True)
|
||||
ws_path = _write_skill(ws_skills, "ws_only", body="# W")
|
||||
builtin = tmp_path / "builtin"
|
||||
bi_path = _write_skill(builtin, "bi_only", body="# B")
|
||||
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
entries = sorted(loader.list_skills(filter_unavailable=False), key=lambda item: item["name"])
|
||||
assert entries == [
|
||||
{"name": "bi_only", "path": str(bi_path), "source": "builtin"},
|
||||
{"name": "ws_only", "path": str(ws_path), "source": "workspace"},
|
||||
]
|
||||
|
||||
|
||||
def test_list_skills_builtin_omitted_when_dir_missing(tmp_path: Path) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
ws_skills = workspace / "skills"
|
||||
ws_skills.mkdir(parents=True)
|
||||
ws_path = _write_skill(ws_skills, "solo", body="# S")
|
||||
missing_builtin = tmp_path / "no_such_builtin"
|
||||
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=missing_builtin)
|
||||
entries = loader.list_skills(filter_unavailable=False)
|
||||
assert entries == [{"name": "solo", "path": str(ws_path), "source": "workspace"}]
|
||||
|
||||
|
||||
def test_list_skills_filter_unavailable_excludes_unmet_bin_requirement(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
skills_root = workspace / "skills"
|
||||
skills_root.mkdir(parents=True)
|
||||
_write_skill(
|
||||
skills_root,
|
||||
"needs_bin",
|
||||
metadata_json={"requires": {"bins": ["nanobot_test_fake_binary"]}},
|
||||
)
|
||||
builtin = tmp_path / "builtin"
|
||||
builtin.mkdir()
|
||||
|
||||
def fake_which(cmd: str) -> str | None:
|
||||
if cmd == "nanobot_test_fake_binary":
|
||||
return None
|
||||
return "/usr/bin/true"
|
||||
|
||||
monkeypatch.setattr("nanobot.agent.skills.shutil.which", fake_which)
|
||||
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
assert loader.list_skills(filter_unavailable=True) == []
|
||||
|
||||
|
||||
def test_list_skills_filter_unavailable_includes_when_bin_requirement_met(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
skills_root = workspace / "skills"
|
||||
skills_root.mkdir(parents=True)
|
||||
skill_path = _write_skill(
|
||||
skills_root,
|
||||
"has_bin",
|
||||
metadata_json={"requires": {"bins": ["nanobot_test_fake_binary"]}},
|
||||
)
|
||||
builtin = tmp_path / "builtin"
|
||||
builtin.mkdir()
|
||||
|
||||
def fake_which(cmd: str) -> str | None:
|
||||
if cmd == "nanobot_test_fake_binary":
|
||||
return "/fake/nanobot_test_fake_binary"
|
||||
return None
|
||||
|
||||
monkeypatch.setattr("nanobot.agent.skills.shutil.which", fake_which)
|
||||
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
entries = loader.list_skills(filter_unavailable=True)
|
||||
assert entries == [
|
||||
{"name": "has_bin", "path": str(skill_path), "source": "workspace"},
|
||||
]
|
||||
|
||||
|
||||
def test_list_skills_filter_unavailable_false_keeps_unmet_requirements(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
skills_root = workspace / "skills"
|
||||
skills_root.mkdir(parents=True)
|
||||
skill_path = _write_skill(
|
||||
skills_root,
|
||||
"blocked",
|
||||
metadata_json={"requires": {"bins": ["nanobot_test_fake_binary"]}},
|
||||
)
|
||||
builtin = tmp_path / "builtin"
|
||||
builtin.mkdir()
|
||||
|
||||
monkeypatch.setattr("nanobot.agent.skills.shutil.which", lambda _cmd: None)
|
||||
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
entries = loader.list_skills(filter_unavailable=False)
|
||||
assert entries == [
|
||||
{"name": "blocked", "path": str(skill_path), "source": "workspace"},
|
||||
]
|
||||
|
||||
|
||||
def test_list_skills_filter_unavailable_excludes_unmet_env_requirement(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
skills_root = workspace / "skills"
|
||||
skills_root.mkdir(parents=True)
|
||||
_write_skill(
|
||||
skills_root,
|
||||
"needs_env",
|
||||
metadata_json={"requires": {"env": ["NANOBOT_SKILLS_TEST_ENV_VAR"]}},
|
||||
)
|
||||
builtin = tmp_path / "builtin"
|
||||
builtin.mkdir()
|
||||
|
||||
monkeypatch.delenv("NANOBOT_SKILLS_TEST_ENV_VAR", raising=False)
|
||||
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
assert loader.list_skills(filter_unavailable=True) == []
|
||||
|
||||
|
||||
def test_list_skills_openclaw_metadata_parsed_for_requirements(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
workspace = tmp_path / "ws"
|
||||
skills_root = workspace / "skills"
|
||||
skills_root.mkdir(parents=True)
|
||||
skill_dir = skills_root / "openclaw_skill"
|
||||
skill_dir.mkdir(parents=True)
|
||||
skill_path = skill_dir / "SKILL.md"
|
||||
oc_payload = json.dumps({"openclaw": {"requires": {"bins": ["nanobot_oc_bin"]}}}, separators=(",", ":"))
|
||||
skill_path.write_text(
|
||||
"\n".join(["---", f"metadata: {oc_payload}", "---", "", "# OC"]),
|
||||
encoding="utf-8",
|
||||
)
|
||||
builtin = tmp_path / "builtin"
|
||||
builtin.mkdir()
|
||||
|
||||
monkeypatch.setattr("nanobot.agent.skills.shutil.which", lambda _cmd: None)
|
||||
|
||||
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
|
||||
assert loader.list_skills(filter_unavailable=True) == []
|
||||
|
||||
monkeypatch.setattr(
|
||||
"nanobot.agent.skills.shutil.which",
|
||||
lambda cmd: "/x" if cmd == "nanobot_oc_bin" else None,
|
||||
)
|
||||
entries = loader.list_skills(filter_unavailable=True)
|
||||
assert entries == [
|
||||
{"name": "openclaw_skill", "path": str(skill_path), "source": "workspace"},
|
||||
]
|
||||
@@ -1,3 +1,6 @@
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
from nanobot.agent.tools import (
|
||||
@@ -546,10 +549,15 @@ async def test_exec_head_tail_truncation() -> None:
|
||||
"""Long output should preserve both head and tail."""
|
||||
tool = ExecTool()
|
||||
# Generate output that exceeds _MAX_OUTPUT (10_000 chars)
|
||||
# Use python to generate output to avoid command line length limits
|
||||
result = await tool.execute(
|
||||
command="python -c \"print('A' * 6000 + '\\n' + 'B' * 6000)\""
|
||||
)
|
||||
# Use current interpreter (PATH may not have `python`). ExecTool uses
|
||||
# create_subprocess_shell: POSIX needs shlex.quote; Windows uses cmd.exe
|
||||
# rules, so list2cmdline is appropriate there.
|
||||
script = "print('A' * 6000 + '\\n' + 'B' * 6000)"
|
||||
if sys.platform == "win32":
|
||||
command = subprocess.list2cmdline([sys.executable, "-c", script])
|
||||
else:
|
||||
command = f"{shlex.quote(sys.executable)} -c {shlex.quote(script)}"
|
||||
result = await tool.execute(command=command)
|
||||
assert "chars truncated" in result
|
||||
# Head portion should start with As
|
||||
assert result.startswith("A")
|
||||
|
||||
Reference in New Issue
Block a user