test/tools/skills/test_skill_search.py
Rama main · réplica de lectura del repositorio original.
# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0
import io
import json
import os
import stat
import tarfile
import textwrap
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
from ag2 import Context
from ag2.events import ToolCallEvent, ToolErrorEvent
from ag2.exceptions import InvalidSkillError, SkillInstallError
from ag2.stream import MemoryStream
from ag2.tools.skills import LocalRuntime, SkillSearchToolkit
from ag2.tools.skills.skill_search.client import SkillsClient
from ag2.tools.skills.skill_search.extractor import extract_skill
from ag2.tools.skills.skill_search.lock import SkillsLock
from ag2.tools.skills.skill_types import SkillMetadata
MONOREPO_SKILL_MD = textwrap.dedent("""\
---
name: vercel-react-best-practices
description: React and Next.js performance optimization guidelines
version: 1.0.0
---
# React Best Practices
Use functional components and hooks.
""")
STANDALONE_SKILL_MD = textwrap.dedent("""\
---
name: last30days
description: Last 30 days analytics script
---
# Last 30 Days
Run last30days.py to get analytics.
""")
def _make_tarball(entries: dict[str, bytes | str]) -> bytes:
"""Build an in-memory .tar.gz archive from a name->content mapping."""
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
for name, content in entries.items():
data = content.encode() if isinstance(content, str) else content
info = tarfile.TarInfo(name=name)
info.size = len(data)
tar.addfile(info, io.BytesIO(data))
return buf.getvalue()
def _monorepo_tarball(skill_id: str = "react-best-practices") -> bytes:
return _make_tarball({
f"owner-repo-abc123/skills/{skill_id}/SKILL.md": MONOREPO_SKILL_MD,
f"owner-repo-abc123/skills/{skill_id}/rules/rule1.md": "# Rule 1\nDo good.",
})
def _standalone_tarball() -> bytes:
return _make_tarball({
"owner-repo-abc123/SKILL.md": STANDALONE_SKILL_MD,
"owner-repo-abc123/scripts/last30days.py": 'print("hello")\n',
})
def _make_meta(tmp_path: Path, name: str = "vercel-react-best-practices") -> SkillMetadata:
"""Build a SkillMetadata for mocked install tests."""
skill_dir = tmp_path / "skills" / name
skill_dir.mkdir(parents=True, exist_ok=True)
return SkillMetadata(
name=name,
description="React and Next.js performance optimization guidelines",
version="1.0.0",
)
def test_extract_skill_monorepo(tmp_path: Path) -> None:
dest = tmp_path / "skills"
dest.mkdir()
tar_path = tmp_path / "skill.tar.gz"
tar_path.write_bytes(_monorepo_tarball())
meta = extract_skill(tar_path, "react-best-practices", dest)
assert meta.name == "vercel-react-best-practices"
assert meta.description == "React and Next.js performance optimization guidelines"
assert meta.version == "1.0.0"
skill_dir = dest / "vercel-react-best-practices"
assert (skill_dir / "SKILL.md").exists()
assert (skill_dir / "rules" / "rule1.md").exists()
def test_extract_skill_standalone(tmp_path: Path) -> None:
dest = tmp_path / "skills"
dest.mkdir()
tar_path = tmp_path / "skill.tar.gz"
tar_path.write_bytes(_standalone_tarball())
meta = extract_skill(tar_path, "", dest)
assert meta.name == "last30days"
assert (dest / "last30days" / "SKILL.md").exists()
assert (dest / "last30days" / "scripts" / "last30days.py").exists()
def test_extract_skill_no_skill_md_raises(tmp_path: Path) -> None:
dest = tmp_path / "skills"
dest.mkdir()
tar_path = tmp_path / "skill.tar.gz"
tar_path.write_bytes(_make_tarball({"owner-repo-abc123/README.md": "# Nothing\n"}))
with pytest.raises(SkillInstallError, match="No SKILL.md found"):
extract_skill(tar_path, "", dest)
def test_extract_skill_excludes_git_dir(tmp_path: Path) -> None:
dest = tmp_path / "skills"
dest.mkdir()
tar_path = tmp_path / "skill.tar.gz"
tar_path.write_bytes(
_make_tarball({
"owner-repo-abc123/SKILL.md": STANDALONE_SKILL_MD,
"owner-repo-abc123/.git/config": "[core]\nbare = false\n",
})
)
extract_skill(tar_path, "", dest)
assert not (dest / "last30days" / ".git").exists()
def test_extract_skill_overwrites_existing(tmp_path: Path) -> None:
dest = tmp_path / "skills"
dest.mkdir()
(dest / "last30days").mkdir()
(dest / "last30days" / "stale.txt").write_text("old")
tar_path = tmp_path / "skill.tar.gz"
tar_path.write_bytes(_standalone_tarball())
extract_skill(tar_path, "", dest)
assert not (dest / "last30days" / "stale.txt").exists()
assert (dest / "last30days" / "SKILL.md").exists()
def test_extract_skill_validates_metadata(tmp_path: Path) -> None:
"""Extracted skill with invalid name format should raise."""
dest = tmp_path / "skills"
dest.mkdir()
bad_skill_md = "---\nname: INVALID_NAME\ndescription: Bad\n---\n"
tar_path = tmp_path / "skill.tar.gz"
tar_path.write_bytes(_make_tarball({"owner-repo-abc123/SKILL.md": bad_skill_md}))
with pytest.raises(InvalidSkillError):
extract_skill(tar_path, "", dest)
@pytest.mark.parametrize("absolute", [False, True], ids=["relative", "absolute"])
def test_extract_skill_rejects_path_name_before_touching_disk(tmp_path: Path, absolute: bool) -> None:
dest = tmp_path / "skills"
dest.mkdir()
victim = tmp_path / "victim"
victim.mkdir()
(victim / "keep.txt").write_text("keep")
name = str(victim) if absolute else "../victim"
skill_md = f"---\nname: {name}\ndescription: Bad\n---\n"
tar_path = tmp_path / "skill.tar.gz"
tar_path.write_bytes(_make_tarball({"owner-repo-abc123/SKILL.md": skill_md}))
with pytest.raises(InvalidSkillError):
extract_skill(tar_path, "", dest)
assert (victim / "keep.txt").read_text() == "keep"
assert not (victim / "SKILL.md").exists()
def test_extract_skill_skips_absolute_member_path(tmp_path: Path) -> None:
dest = tmp_path / "skills"
dest.mkdir()
victim = tmp_path / "victim"
victim.mkdir()
tar_path = tmp_path / "skill.tar.gz"
tar_path.write_bytes(
_make_tarball({
"owner-repo-abc123/SKILL.md": STANDALONE_SKILL_MD,
f"owner-repo-abc123/{victim}/pwned.txt": "pwned",
})
)
extract_skill(tar_path, "", dest)
assert not (victim / "pwned.txt").exists()
assert (dest / "last30days" / "SKILL.md").exists()
@pytest.mark.asyncio
@pytest.mark.skipif(os.name == "nt", reason="POSIX executable permissions")
@pytest.mark.parametrize("helper_mode", [0o755, 0o4755])
async def test_extract_skill_keeps_executable_bits_for_bundled_helpers(tmp_path: Path, helper_mode: int) -> None:
archive = tmp_path / "skill.tar.gz"
entries = [
("SKILL.md", b"---\nname: helper-skill\ndescription: Run a bundled helper\n---\n", 0o644),
("scripts/run.sh", b"#!/bin/sh\n./helper\n", 0o755),
("scripts/helper", b"#!/bin/sh\nprintf 'helper output'\n", helper_mode),
("references/note.txt", b"Not executable", 0o644),
]
with tarfile.open(archive, "w:gz") as tar:
for name, data, mode in entries:
member = tarfile.TarInfo(f"repo-main/{name}")
member.size = len(data)
member.mode = mode
tar.addfile(member, io.BytesIO(data))
install_dir = tmp_path / "skills"
extract_skill(archive, "", install_dir)
runtime = LocalRuntime(install_dir)
output = await runtime.execute("helper-skill", "run.sh", Context(stream=MemoryStream()))
assert output == "helper output"
skill_dir = install_dir / "helper-skill"
helper_mode_installed = stat.S_IMODE((skill_dir / "scripts/helper").stat().st_mode)
assert helper_mode_installed & 0o111 == 0o111
assert helper_mode_installed & 0o7000 == 0
assert (skill_dir / "references/note.txt").stat().st_mode & 0o111 == 0
def test_local_runtime_install_path_traversal_blocked(tmp_path: Path) -> None:
install_dir = tmp_path / "skills"
install_dir.mkdir()
source = tmp_path / "source"
source.mkdir()
with pytest.raises(ValueError, match="path traversal"):
LocalRuntime(dir=install_dir).install(source, "../outside")
assert not (tmp_path / "outside").exists()
@pytest.mark.asyncio
async def test_search_skills_formats_output(tmp_path: Path) -> None:
skills_data = [
{
"skillId": "react-best-practices",
"name": "vercel-react-best-practices",
"installs": 229780,
"source": "vercel-labs/agent-skills",
},
{"skillId": "nextjs-patterns", "name": "nextjs-patterns", "installs": 5000, "source": "some-user/nextjs-skill"},
]
with patch.object(SkillsClient, "search", AsyncMock(return_value=skills_data)):
toolkit = SkillSearchToolkit(runtime=tmp_path / "skills")
result = await toolkit.search_skills().model.call(query="react")
assert 'Found 2 skill(s) for "react"' in result
assert "vercel-react-best-practices" in result
assert "229,780 installs" in result
assert 'install_skill("vercel-labs/agent-skills/react-best-practices")' in result
@pytest.mark.asyncio
async def test_search_skills_no_results(tmp_path: Path) -> None:
with patch.object(SkillsClient, "search", AsyncMock(return_value=[])):
toolkit = SkillSearchToolkit(runtime=tmp_path / "skills")
result = await toolkit.search_skills().model.call(query="xyzzy-nonexistent")
assert "No skills found" in result
@pytest.mark.asyncio
async def test_search_skills_network_error(tmp_path: Path) -> None:
with patch.object(SkillsClient, "search", AsyncMock(side_effect=Exception("connection refused"))):
toolkit = SkillSearchToolkit(runtime=tmp_path / "skills")
result = await toolkit.search_skills().model.call(query="react")
assert "Error searching skills.sh" in result
assert "connection refused" in result
@pytest.mark.asyncio
async def test_install_skill_monorepo(tmp_path: Path) -> None:
meta = _make_meta(tmp_path)
with patch.object(SkillsClient, "download_skill", AsyncMock(return_value=(meta, "abc123hash"))):
toolkit = SkillSearchToolkit(runtime=tmp_path / "skills")
result = await toolkit.install_skill().model.call(skill_id="vercel-labs/agent-skills/react-best-practices")
assert "Installed: vercel-react-best-practices" in result
assert "Description:" in result
assert 'load_skill("vercel-react-best-practices")' in result
@pytest.mark.asyncio
async def test_install_skill_standalone(tmp_path: Path) -> None:
meta = _make_meta(tmp_path, name="last30days")
with patch.object(SkillsClient, "download_skill", AsyncMock(return_value=(meta, "def456hash"))):
toolkit = SkillSearchToolkit(runtime=tmp_path / "skills")
result = await toolkit.install_skill().model.call(skill_id="mvanhorn/last30days-skill")
assert "Installed: last30days" in result
@pytest.mark.asyncio
async def test_install_skill_records_hash(tmp_path: Path) -> None:
"""install_skill should write hash to skills-lock.json."""
install_dir = tmp_path / "skills"
meta = _make_meta(tmp_path)
with patch.object(SkillsClient, "download_skill", AsyncMock(return_value=(meta, "abc123hash"))):
toolkit = SkillSearchToolkit(runtime=install_dir)
await toolkit.install_skill().model.call(skill_id="vercel-labs/agent-skills/react-best-practices")
lock_path = install_dir / "skills-lock.json"
assert lock_path.exists()
lock_data = json.loads(lock_path.read_text())
assert lock_data == {
"version": 1,
"skills": {
"vercel-react-best-practices": {
"source": "vercel-labs/agent-skills",
"sourceType": "github",
"computedHash": "abc123hash",
}
},
}
@pytest.mark.asyncio
async def test_install_skill_rate_limit(tmp_path: Path) -> None:
err = RuntimeError("GitHub rate limit exceeded. Set GITHUB_TOKEN")
with patch.object(SkillsClient, "download_skill", AsyncMock(side_effect=err)):
toolkit = SkillSearchToolkit(runtime=tmp_path / "skills")
result = await toolkit.install_skill().model.call(skill_id="some/repo/skill")
assert "rate limit" in result.lower()
@pytest.mark.asyncio
async def test_install_skill_not_found(tmp_path: Path) -> None:
err = RuntimeError("Skill not found: no-such/repo")
with patch.object(SkillsClient, "download_skill", AsyncMock(side_effect=err)):
toolkit = SkillSearchToolkit(runtime=tmp_path / "skills")
result = await toolkit.install_skill().model.call(skill_id="no-such/repo/skill")
assert "not found" in result.lower()
@pytest.mark.asyncio
async def test_install_skill_invalid_id(tmp_path: Path) -> None:
toolkit = SkillSearchToolkit(runtime=tmp_path / "skills")
result = await toolkit.install_skill().model.call(skill_id="invalid")
assert "Invalid skill_id format" in result
@pytest.mark.asyncio
async def test_remove_skill_success(tmp_path: Path) -> None:
install_dir = tmp_path / "skills"
skill_dir = install_dir / "my-skill"
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text("---\nname: my-skill\n---\n")
toolkit = SkillSearchToolkit(runtime=install_dir)
result = await toolkit.remove_skill().model.call(name="my-skill")
assert result == "Removed: my-skill"
assert not skill_dir.exists()
@pytest.mark.asyncio
async def test_remove_skill_cleans_lock_file(tmp_path: Path) -> None:
"""remove_skill should remove entry from skills-lock.json."""
install_dir = tmp_path / "skills"
skill_dir = install_dir / "my-skill"
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text("---\nname: my-skill\n---\n")
# Pre-populate lock file
lock_path = install_dir / "skills-lock.json"
lock_data = {"version": 1, "skills": {"my-skill": {"source": "x/y", "sourceType": "github", "computedHash": "aaa"}}}
lock_path.write_text(json.dumps(lock_data))
toolkit = SkillSearchToolkit(runtime=install_dir)
await toolkit.remove_skill().model.call(name="my-skill")
updated = json.loads(lock_path.read_text())
assert "my-skill" not in updated["skills"]
@pytest.mark.asyncio
async def test_remove_skill_not_found(tmp_path: Path) -> None:
install_dir = tmp_path / "skills"
install_dir.mkdir()
toolkit = SkillSearchToolkit(runtime=install_dir)
result = await toolkit.remove_skill().model.call(name="nonexistent")
assert "Cannot remove" in result
@pytest.mark.asyncio
async def test_remove_skill_path_traversal_blocked(tmp_path: Path) -> None:
install_dir = tmp_path / "skills"
install_dir.mkdir()
outside = tmp_path / "secret"
outside.mkdir()
toolkit = SkillSearchToolkit(runtime=install_dir)
result = await toolkit.remove_skill().model.call(name="../secret")
assert "Cannot remove" in result
assert outside.exists()
def test_lock_record_and_read(tmp_path: Path) -> None:
lock = SkillsLock(tmp_path / "skills-lock.json")
lock.record("my-skill", "owner/repo", "abc123")
data = lock.read()
assert data == {
"version": 1,
"skills": {
"my-skill": {
"source": "owner/repo",
"sourceType": "github",
"computedHash": "abc123",
}
},
}
def test_lock_remove(tmp_path: Path) -> None:
lock = SkillsLock(tmp_path / "skills-lock.json")
lock.record("skill-a", "a/b", "hash1")
lock.record("skill-b", "c/d", "hash2")
lock.remove("skill-a")
data = lock.read()
assert "skill-a" not in data["skills"]
assert "skill-b" in data["skills"]
def test_lock_get_hash(tmp_path: Path) -> None:
lock = SkillsLock(tmp_path / "skills-lock.json")
lock.record("my-skill", "owner/repo", "abc123")
assert lock.get_hash("my-skill") == "abc123"
assert lock.get_hash("nonexistent") is None
def test_lock_read_nonexistent(tmp_path: Path) -> None:
lock = SkillsLock(tmp_path / "no-such-file.json")
data = lock.read()
assert data == {"version": 1, "skills": {}}
@pytest.mark.asyncio
async def test_toolkit_exposes_all_tools(tmp_path: Path, context: Context) -> None:
toolkit = SkillSearchToolkit(runtime=tmp_path / "skills")
schemas = list(await toolkit.schemas(context))
names = {s.function.name for s in schemas} # type: ignore[union-attr]
assert names == {
"search_skills",
"install_skill",
"remove_skill",
"list_skills",
"load_skill",
"read_skill_resource",
"run_skill_script",
}
@pytest.mark.asyncio
async def test_load_skill_name_not_pinned_to_construction_time_skills(tmp_path: Path, context: Context) -> None:
# SkillSearchToolkit installs skills at runtime, so the activation tools must
# accept ANY skill name. Pinning `name` to a Literal of the skills present at
# construction would reject a freshly installed skill.
skill_dir = tmp_path / "skills" / "existing-skill"
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text("---\nname: existing-skill\ndescription: x\n---\n# x\n")
toolkit = SkillSearchToolkit(LocalRuntime(dir=tmp_path / "skills"))
[schema] = await toolkit.load_skill().schemas(context)
name_prop = schema.function.parameters["properties"]["name"] # type: ignore[union-attr]
assert name_prop["type"] == "string"
# No pinning: neither a Literal enum (2+ skills) nor a const (1 skill).
assert "enum" not in name_prop
assert "const" not in name_prop
@pytest.mark.asyncio
async def test_load_skill_accepts_skill_installed_after_construction(tmp_path: Path, context: Context) -> None:
# The end-to-end payoff: a skill that appears after the toolkit is built (as
# install_skill would do) is loadable — its name is not rejected by a pinned
# `name` constraint baked in at construction.
skills_dir = tmp_path / "skills"
a = skills_dir / "skill-a"
a.mkdir(parents=True)
(a / "SKILL.md").write_text("---\nname: skill-a\ndescription: a\n---\n# A\n")
toolkit = SkillSearchToolkit(LocalRuntime(dir=skills_dir))
load_tool = toolkit.load_skill()
# Simulate install_skill: a new skill lands in the runtime, cache invalidated.
b = skills_dir / "skill-b"
b.mkdir()
(b / "SKILL.md").write_text("---\nname: skill-b\ndescription: b\n---\n# Skill B body\n")
toolkit.runtimes[0].invalidate()
event = ToolCallEvent(name="load_skill", arguments=json.dumps({"name": "skill-b"}))
result = await load_tool(event, context)
assert not isinstance(result, ToolErrorEvent)
assert "Skill B body" in result.result.parts[0].content
@pytest.mark.asyncio
async def test_toolkit_individual_tools_accessible(tmp_path: Path, context: Context) -> None:
toolkit = SkillSearchToolkit(runtime=LocalRuntime(dir=tmp_path / "skills"))
tools = (
"search_skills",
"install_skill",
"remove_skill",
"list_skills",
"load_skill",
"read_skill_resource",
"run_skill_script",
)
for attr in tools:
[schema] = await getattr(toolkit, attr)().schemas(context)
assert schema.function.name == attr # type: ignore[union-attr]