5dbe2d9d73
* fix(run_agent): ensure _fire_first_delta() is called for tool generation events Added calls to _fire_first_delta() in the AIAgent class to improve the handling of tool generation events, ensuring timely notifications during the processing of function calls and tool usage. * fix(run_agent): improve timeout handling for chat completions Enhanced the timeout configuration for chat completions in the AIAgent class by introducing customizable connection, read, and write timeouts using environment variables. This ensures more robust handling of API requests during streaming operations. * fix(run_agent): reduce default stream read timeout for chat completions Updated the default stream read timeout from 120 seconds to 60 seconds in the AIAgent class, enhancing the timeout configuration for chat completions. This change aims to improve responsiveness during streaming operations. * fix(run_agent): enhance streaming error handling and retry logic Improved the error handling and retry mechanism for streaming requests in the AIAgent class. Introduced a configurable maximum number of stream retries and refined the handling of transient network errors, allowing for retries with fresh connections. Non-transient errors now trigger a fallback to non-streaming only when appropriate, ensuring better resilience during API interactions. * fix: skills-sh install fails for deeply nested repo structures Skills in repos with deep directory nesting (e.g. cli-tool/components/skills/development/senior-backend/) could not be installed because the candidate path generation and shallow root-dir scan never reached them. Added GitHubSource._find_skill_in_repo_tree() which uses the GitHub Trees API to recursively search the entire repo tree in a single API call. This is used as a final fallback in SkillsShSource._discover_identifier() when the standard candidate paths and shallow scan both fail. Fixes installation of skills from repos like davila7/claude-code-templates where skills are nested 4+ levels deep. Reported by user Samuraixheart.
1042 lines
42 KiB
Python
1042 lines
42 KiB
Python
"""Tests for tools/skills_hub.py — source adapters, lock file, taps, dedup logic."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from tools.skills_hub import (
|
|
GitHubAuth,
|
|
GitHubSource,
|
|
LobeHubSource,
|
|
SkillsShSource,
|
|
WellKnownSkillSource,
|
|
OptionalSkillSource,
|
|
SkillMeta,
|
|
SkillBundle,
|
|
HubLockFile,
|
|
TapsManager,
|
|
bundle_content_hash,
|
|
check_for_skill_updates,
|
|
create_source_router,
|
|
unified_search,
|
|
append_audit_log,
|
|
_skill_meta_to_dict,
|
|
quarantine_bundle,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GitHubSource._parse_frontmatter_quick
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestParseFrontmatterQuick:
|
|
def test_valid_frontmatter(self):
|
|
content = "---\nname: test-skill\ndescription: A test.\n---\n\n# Body\n"
|
|
fm = GitHubSource._parse_frontmatter_quick(content)
|
|
assert fm["name"] == "test-skill"
|
|
assert fm["description"] == "A test."
|
|
|
|
def test_no_frontmatter(self):
|
|
content = "# Just a heading\nSome body text.\n"
|
|
fm = GitHubSource._parse_frontmatter_quick(content)
|
|
assert fm == {}
|
|
|
|
def test_no_closing_delimiter(self):
|
|
content = "---\nname: test\ndescription: desc\nno closing here\n"
|
|
fm = GitHubSource._parse_frontmatter_quick(content)
|
|
assert fm == {}
|
|
|
|
def test_empty_content(self):
|
|
fm = GitHubSource._parse_frontmatter_quick("")
|
|
assert fm == {}
|
|
|
|
def test_nested_yaml(self):
|
|
content = "---\nname: test\nmetadata:\n hermes:\n tags: [a, b]\n---\n\nBody.\n"
|
|
fm = GitHubSource._parse_frontmatter_quick(content)
|
|
assert fm["metadata"]["hermes"]["tags"] == ["a", "b"]
|
|
|
|
def test_invalid_yaml_returns_empty(self):
|
|
content = "---\n: : : invalid{{\n---\n\nBody.\n"
|
|
fm = GitHubSource._parse_frontmatter_quick(content)
|
|
assert fm == {}
|
|
|
|
def test_non_dict_yaml_returns_empty(self):
|
|
content = "---\n- just a list\n- of items\n---\n\nBody.\n"
|
|
fm = GitHubSource._parse_frontmatter_quick(content)
|
|
assert fm == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GitHubSource.trust_level_for
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTrustLevelFor:
|
|
def _source(self):
|
|
auth = MagicMock(spec=GitHubAuth)
|
|
return GitHubSource(auth=auth)
|
|
|
|
def test_trusted_repo(self):
|
|
src = self._source()
|
|
# TRUSTED_REPOS is imported from skills_guard, test with known trusted repo
|
|
from tools.skills_guard import TRUSTED_REPOS
|
|
if TRUSTED_REPOS:
|
|
repo = next(iter(TRUSTED_REPOS))
|
|
assert src.trust_level_for(f"{repo}/some-skill") == "trusted"
|
|
|
|
def test_community_repo(self):
|
|
src = self._source()
|
|
assert src.trust_level_for("random-user/random-repo/skill") == "community"
|
|
|
|
def test_short_identifier(self):
|
|
src = self._source()
|
|
assert src.trust_level_for("no-slash") == "community"
|
|
|
|
def test_two_part_identifier(self):
|
|
src = self._source()
|
|
result = src.trust_level_for("owner/repo")
|
|
# No path part — still resolves repo correctly
|
|
assert result in ("trusted", "community")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SkillsShSource
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSkillsShSource:
|
|
def _source(self):
|
|
auth = MagicMock(spec=GitHubAuth)
|
|
return SkillsShSource(auth=auth)
|
|
|
|
@patch("tools.skills_hub._write_index_cache")
|
|
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
|
@patch("tools.skills_hub.httpx.get")
|
|
def test_search_maps_skills_sh_results_to_prefixed_identifiers(self, mock_get, _mock_read_cache, _mock_write_cache):
|
|
mock_get.return_value = MagicMock(
|
|
status_code=200,
|
|
json=lambda: {
|
|
"skills": [
|
|
{
|
|
"id": "vercel-labs/agent-skills/vercel-react-best-practices",
|
|
"skillId": "vercel-react-best-practices",
|
|
"name": "vercel-react-best-practices",
|
|
"installs": 207679,
|
|
"source": "vercel-labs/agent-skills",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
results = self._source().search("react", limit=5)
|
|
|
|
assert len(results) == 1
|
|
assert results[0].source == "skills.sh"
|
|
assert results[0].identifier == "skills-sh/vercel-labs/agent-skills/vercel-react-best-practices"
|
|
assert "skills.sh" in results[0].description
|
|
assert results[0].repo == "vercel-labs/agent-skills"
|
|
assert results[0].path == "vercel-react-best-practices"
|
|
assert results[0].extra["installs"] == 207679
|
|
|
|
@patch("tools.skills_hub._write_index_cache")
|
|
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
|
@patch("tools.skills_hub.httpx.get")
|
|
def test_empty_search_uses_featured_homepage_links(self, mock_get, _mock_read_cache, _mock_write_cache):
|
|
mock_get.return_value = MagicMock(
|
|
status_code=200,
|
|
text='''
|
|
<a href="/vercel-labs/agent-skills/vercel-react-best-practices">React</a>
|
|
<a href="/anthropics/skills/pdf">PDF</a>
|
|
<a href="/vercel-labs/agent-skills/vercel-react-best-practices">React again</a>
|
|
''',
|
|
)
|
|
|
|
results = self._source().search("", limit=10)
|
|
|
|
assert [r.identifier for r in results] == [
|
|
"skills-sh/vercel-labs/agent-skills/vercel-react-best-practices",
|
|
"skills-sh/anthropics/skills/pdf",
|
|
]
|
|
assert all(r.source == "skills.sh" for r in results)
|
|
|
|
@patch.object(GitHubSource, "fetch")
|
|
def test_fetch_delegates_to_github_source_and_relabels_bundle(self, mock_fetch):
|
|
mock_fetch.return_value = SkillBundle(
|
|
name="vercel-react-best-practices",
|
|
files={"SKILL.md": "# Test"},
|
|
source="github",
|
|
identifier="vercel-labs/agent-skills/vercel-react-best-practices",
|
|
trust_level="community",
|
|
)
|
|
|
|
bundle = self._source().fetch("skills-sh/vercel-labs/agent-skills/vercel-react-best-practices")
|
|
|
|
assert bundle is not None
|
|
assert bundle.source == "skills.sh"
|
|
assert bundle.identifier == "skills-sh/vercel-labs/agent-skills/vercel-react-best-practices"
|
|
mock_fetch.assert_called_once_with("vercel-labs/agent-skills/vercel-react-best-practices")
|
|
|
|
@patch("tools.skills_hub._write_index_cache")
|
|
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
|
@patch("tools.skills_hub.httpx.get")
|
|
@patch.object(GitHubSource, "inspect")
|
|
def test_inspect_delegates_to_github_source_and_relabels_meta(self, mock_inspect, mock_get, _mock_read_cache, _mock_write_cache):
|
|
mock_inspect.return_value = SkillMeta(
|
|
name="vercel-react-best-practices",
|
|
description="React rules",
|
|
source="github",
|
|
identifier="vercel-labs/agent-skills/vercel-react-best-practices",
|
|
trust_level="community",
|
|
repo="vercel-labs/agent-skills",
|
|
path="vercel-react-best-practices",
|
|
)
|
|
mock_get.return_value = MagicMock(
|
|
status_code=200,
|
|
text='''
|
|
<h1>vercel-react-best-practices</h1>
|
|
<code>$ npx skills add https://github.com/vercel-labs/agent-skills --skill vercel-react-best-practices</code>
|
|
<div class="prose"><h1>Vercel React Best Practices</h1><p>React rules.</p></div>
|
|
<a href="/vercel-labs/agent-skills/vercel-react-best-practices/security/socket">Socket</a> Pass
|
|
<a href="/vercel-labs/agent-skills/vercel-react-best-practices/security/snyk">Snyk</a> Pass
|
|
''',
|
|
)
|
|
|
|
meta = self._source().inspect("skills-sh/vercel-labs/agent-skills/vercel-react-best-practices")
|
|
|
|
assert meta is not None
|
|
assert meta.source == "skills.sh"
|
|
assert meta.identifier == "skills-sh/vercel-labs/agent-skills/vercel-react-best-practices"
|
|
assert meta.extra["install_command"].endswith("--skill vercel-react-best-practices")
|
|
assert meta.extra["security_audits"]["socket"] == "Pass"
|
|
mock_inspect.assert_called_once_with("vercel-labs/agent-skills/vercel-react-best-practices")
|
|
|
|
@patch.object(GitHubSource, "_list_skills_in_repo")
|
|
@patch.object(GitHubSource, "inspect")
|
|
def test_inspect_falls_back_to_repo_skill_catalog_when_slug_differs(self, mock_inspect, mock_list_skills):
|
|
resolved = SkillMeta(
|
|
name="vercel-react-best-practices",
|
|
description="React rules",
|
|
source="github",
|
|
identifier="vercel-labs/agent-skills/skills/react-best-practices",
|
|
trust_level="community",
|
|
repo="vercel-labs/agent-skills",
|
|
path="skills/react-best-practices",
|
|
)
|
|
mock_inspect.side_effect = lambda identifier: resolved if identifier == resolved.identifier else None
|
|
mock_list_skills.return_value = [resolved]
|
|
|
|
meta = self._source().inspect("skills-sh/vercel-labs/agent-skills/vercel-react-best-practices")
|
|
|
|
assert meta is not None
|
|
assert meta.identifier == "skills-sh/vercel-labs/agent-skills/vercel-react-best-practices"
|
|
assert mock_list_skills.called
|
|
|
|
@patch("tools.skills_hub._write_index_cache")
|
|
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
|
@patch("tools.skills_hub.httpx.get")
|
|
@patch.object(GitHubSource, "_list_skills_in_repo")
|
|
@patch.object(GitHubSource, "inspect")
|
|
def test_inspect_uses_detail_page_to_resolve_alias_skill(self, mock_inspect, mock_list_skills, mock_get, _mock_read_cache, _mock_write_cache):
|
|
resolved = SkillMeta(
|
|
name="react",
|
|
description="React renderer",
|
|
source="github",
|
|
identifier="vercel-labs/json-render/skills/react",
|
|
trust_level="community",
|
|
repo="vercel-labs/json-render",
|
|
path="skills/react",
|
|
)
|
|
mock_inspect.side_effect = lambda identifier: resolved if identifier == resolved.identifier else None
|
|
mock_list_skills.return_value = [resolved]
|
|
mock_get.return_value = MagicMock(
|
|
status_code=200,
|
|
text='''
|
|
<h1>json-render-react</h1>
|
|
<code>$ npx skills add https://github.com/vercel-labs/json-render --skill json-render-react</code>
|
|
<div class="prose"><h1>@json-render/react</h1><p>React renderer.</p></div>
|
|
''',
|
|
)
|
|
|
|
meta = self._source().inspect("skills-sh/vercel-labs/json-render/json-render-react")
|
|
|
|
assert meta is not None
|
|
assert meta.identifier == "skills-sh/vercel-labs/json-render/json-render-react"
|
|
assert meta.path == "skills/react"
|
|
assert mock_get.called
|
|
|
|
@patch("tools.skills_hub._write_index_cache")
|
|
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
|
@patch("tools.skills_hub.httpx.get")
|
|
@patch.object(GitHubSource, "_list_skills_in_repo")
|
|
@patch.object(GitHubSource, "fetch")
|
|
def test_fetch_uses_detail_page_to_resolve_alias_skill(self, mock_fetch, mock_list_skills, mock_get, _mock_read_cache, _mock_write_cache):
|
|
resolved_meta = SkillMeta(
|
|
name="react",
|
|
description="React renderer",
|
|
source="github",
|
|
identifier="vercel-labs/json-render/skills/react",
|
|
trust_level="community",
|
|
repo="vercel-labs/json-render",
|
|
path="skills/react",
|
|
)
|
|
resolved_bundle = SkillBundle(
|
|
name="react",
|
|
files={"SKILL.md": "# react"},
|
|
source="github",
|
|
identifier="vercel-labs/json-render/skills/react",
|
|
trust_level="community",
|
|
)
|
|
mock_fetch.side_effect = lambda identifier: resolved_bundle if identifier == resolved_bundle.identifier else None
|
|
mock_list_skills.return_value = [resolved_meta]
|
|
mock_get.return_value = MagicMock(
|
|
status_code=200,
|
|
text='''
|
|
<h1>json-render-react</h1>
|
|
<code>$ npx skills add https://github.com/vercel-labs/json-render --skill json-render-react</code>
|
|
<div class="prose"><h1>@json-render/react</h1><p>React renderer.</p></div>
|
|
''',
|
|
)
|
|
|
|
bundle = self._source().fetch("skills-sh/vercel-labs/json-render/json-render-react")
|
|
|
|
assert bundle is not None
|
|
assert bundle.identifier == "skills-sh/vercel-labs/json-render/json-render-react"
|
|
assert bundle.files["SKILL.md"] == "# react"
|
|
assert mock_get.called
|
|
|
|
@patch("tools.skills_hub._write_index_cache")
|
|
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
|
@patch("tools.skills_hub.httpx.get")
|
|
@patch.object(GitHubSource, "fetch")
|
|
def test_fetch_falls_back_to_tree_search_for_deeply_nested_skills(
|
|
self, mock_fetch, mock_get, _mock_read_cache, _mock_write_cache,
|
|
):
|
|
"""Skills in deeply nested dirs (e.g. cli-tool/components/skills/dev/my-skill/)
|
|
are found via the GitHub Trees API when candidate paths and shallow scan fail."""
|
|
tree_entries = [
|
|
{"path": "README.md", "type": "blob"},
|
|
{"path": "cli-tool/components/skills/development/my-skill/SKILL.md", "type": "blob"},
|
|
{"path": "cli-tool/components/skills/development/other-skill/SKILL.md", "type": "blob"},
|
|
]
|
|
|
|
def _httpx_get_side_effect(url, **kwargs):
|
|
resp = MagicMock()
|
|
if "/api/search" in url:
|
|
resp.status_code = 404
|
|
return resp
|
|
if url.endswith("/contents/"):
|
|
# Root listing for shallow scan — return empty so it falls through
|
|
resp.status_code = 200
|
|
resp.json = lambda: []
|
|
return resp
|
|
if "/contents/" in url:
|
|
# All contents API calls fail (candidate paths miss)
|
|
resp.status_code = 404
|
|
return resp
|
|
if url.endswith("owner/repo"):
|
|
# Repo info → default branch
|
|
resp.status_code = 200
|
|
resp.json = lambda: {"default_branch": "main"}
|
|
return resp
|
|
if "/git/trees/main" in url:
|
|
resp.status_code = 200
|
|
resp.json = lambda: {"tree": tree_entries}
|
|
return resp
|
|
# skills.sh detail page
|
|
resp.status_code = 200
|
|
resp.text = "<h1>my-skill</h1>"
|
|
return resp
|
|
|
|
mock_get.side_effect = _httpx_get_side_effect
|
|
|
|
resolved_bundle = SkillBundle(
|
|
name="my-skill",
|
|
files={"SKILL.md": "# My Skill"},
|
|
source="github",
|
|
identifier="owner/repo/cli-tool/components/skills/development/my-skill",
|
|
trust_level="community",
|
|
)
|
|
mock_fetch.side_effect = lambda ident: resolved_bundle if "cli-tool/components" in ident else None
|
|
|
|
bundle = self._source().fetch("skills-sh/owner/repo/my-skill")
|
|
|
|
assert bundle is not None
|
|
assert bundle.source == "skills.sh"
|
|
assert bundle.files["SKILL.md"] == "# My Skill"
|
|
# Verify the tree-resolved identifier was used for the final GitHub fetch
|
|
mock_fetch.assert_any_call("owner/repo/cli-tool/components/skills/development/my-skill")
|
|
|
|
|
|
class TestFindSkillInRepoTree:
|
|
"""Tests for GitHubSource._find_skill_in_repo_tree."""
|
|
|
|
def _source(self):
|
|
auth = MagicMock(spec=GitHubAuth)
|
|
auth.get_headers.return_value = {"Accept": "application/vnd.github.v3+json"}
|
|
return GitHubSource(auth=auth)
|
|
|
|
@patch("tools.skills_hub.httpx.get")
|
|
def test_finds_deeply_nested_skill(self, mock_get):
|
|
tree_entries = [
|
|
{"path": "README.md", "type": "blob"},
|
|
{"path": "cli-tool/components/skills/development/senior-backend/SKILL.md", "type": "blob"},
|
|
{"path": "cli-tool/components/skills/development/other/SKILL.md", "type": "blob"},
|
|
]
|
|
|
|
def _side_effect(url, **kwargs):
|
|
resp = MagicMock()
|
|
if url.endswith("/davila7/claude-code-templates"):
|
|
resp.status_code = 200
|
|
resp.json = lambda: {"default_branch": "main"}
|
|
elif "/git/trees/main" in url:
|
|
resp.status_code = 200
|
|
resp.json = lambda: {"tree": tree_entries}
|
|
else:
|
|
resp.status_code = 404
|
|
return resp
|
|
|
|
mock_get.side_effect = _side_effect
|
|
|
|
result = self._source()._find_skill_in_repo_tree("davila7/claude-code-templates", "senior-backend")
|
|
assert result == "davila7/claude-code-templates/cli-tool/components/skills/development/senior-backend"
|
|
|
|
@patch("tools.skills_hub.httpx.get")
|
|
def test_finds_root_level_skill(self, mock_get):
|
|
tree_entries = [
|
|
{"path": "my-skill/SKILL.md", "type": "blob"},
|
|
]
|
|
|
|
def _side_effect(url, **kwargs):
|
|
resp = MagicMock()
|
|
if "/contents" not in url and "/git/" not in url:
|
|
resp.status_code = 200
|
|
resp.json = lambda: {"default_branch": "main"}
|
|
elif "/git/trees/main" in url:
|
|
resp.status_code = 200
|
|
resp.json = lambda: {"tree": tree_entries}
|
|
else:
|
|
resp.status_code = 404
|
|
return resp
|
|
|
|
mock_get.side_effect = _side_effect
|
|
|
|
result = self._source()._find_skill_in_repo_tree("owner/repo", "my-skill")
|
|
assert result == "owner/repo/my-skill"
|
|
|
|
@patch("tools.skills_hub.httpx.get")
|
|
def test_returns_none_when_skill_not_found(self, mock_get):
|
|
tree_entries = [
|
|
{"path": "other-skill/SKILL.md", "type": "blob"},
|
|
]
|
|
|
|
def _side_effect(url, **kwargs):
|
|
resp = MagicMock()
|
|
if "/contents" not in url and "/git/" not in url:
|
|
resp.status_code = 200
|
|
resp.json = lambda: {"default_branch": "main"}
|
|
elif "/git/trees/main" in url:
|
|
resp.status_code = 200
|
|
resp.json = lambda: {"tree": tree_entries}
|
|
else:
|
|
resp.status_code = 404
|
|
return resp
|
|
|
|
mock_get.side_effect = _side_effect
|
|
|
|
result = self._source()._find_skill_in_repo_tree("owner/repo", "nonexistent")
|
|
assert result is None
|
|
|
|
@patch("tools.skills_hub.httpx.get")
|
|
def test_returns_none_when_repo_api_fails(self, mock_get):
|
|
mock_get.return_value = MagicMock(status_code=404)
|
|
result = self._source()._find_skill_in_repo_tree("owner/repo", "my-skill")
|
|
assert result is None
|
|
|
|
|
|
class TestWellKnownSkillSource:
|
|
def _source(self):
|
|
return WellKnownSkillSource()
|
|
|
|
@patch("tools.skills_hub._write_index_cache")
|
|
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
|
@patch("tools.skills_hub.httpx.get")
|
|
def test_search_reads_index_from_well_known_url(self, mock_get, _mock_read_cache, _mock_write_cache):
|
|
mock_get.return_value = MagicMock(
|
|
status_code=200,
|
|
json=lambda: {
|
|
"skills": [
|
|
{"name": "git-workflow", "description": "Git rules", "files": ["SKILL.md"]},
|
|
{"name": "code-review", "description": "Review code", "files": ["SKILL.md", "references/checklist.md"]},
|
|
]
|
|
},
|
|
)
|
|
|
|
results = self._source().search("https://example.com/.well-known/skills/index.json", limit=10)
|
|
|
|
assert [r.identifier for r in results] == [
|
|
"well-known:https://example.com/.well-known/skills/git-workflow",
|
|
"well-known:https://example.com/.well-known/skills/code-review",
|
|
]
|
|
assert all(r.source == "well-known" for r in results)
|
|
|
|
@patch("tools.skills_hub._write_index_cache")
|
|
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
|
@patch("tools.skills_hub.httpx.get")
|
|
def test_search_accepts_domain_root_and_resolves_index(self, mock_get, _mock_read_cache, _mock_write_cache):
|
|
mock_get.return_value = MagicMock(
|
|
status_code=200,
|
|
json=lambda: {"skills": [{"name": "git-workflow", "description": "Git rules", "files": ["SKILL.md"]}]},
|
|
)
|
|
|
|
results = self._source().search("https://example.com", limit=10)
|
|
|
|
assert len(results) == 1
|
|
called_url = mock_get.call_args.args[0]
|
|
assert called_url == "https://example.com/.well-known/skills/index.json"
|
|
|
|
@patch("tools.skills_hub._write_index_cache")
|
|
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
|
@patch("tools.skills_hub.httpx.get")
|
|
def test_inspect_fetches_skill_md_from_well_known_endpoint(self, mock_get, _mock_read_cache, _mock_write_cache):
|
|
def fake_get(url, *args, **kwargs):
|
|
if url.endswith("/index.json"):
|
|
return MagicMock(status_code=200, json=lambda: {
|
|
"skills": [{"name": "git-workflow", "description": "Git rules", "files": ["SKILL.md"]}]
|
|
})
|
|
if url.endswith("/git-workflow/SKILL.md"):
|
|
return MagicMock(status_code=200, text="---\nname: git-workflow\ndescription: Git rules\n---\n\n# Git Workflow\n")
|
|
raise AssertionError(url)
|
|
|
|
mock_get.side_effect = fake_get
|
|
|
|
meta = self._source().inspect("well-known:https://example.com/.well-known/skills/git-workflow")
|
|
|
|
assert meta is not None
|
|
assert meta.name == "git-workflow"
|
|
assert meta.source == "well-known"
|
|
assert meta.extra["base_url"] == "https://example.com/.well-known/skills"
|
|
|
|
@patch("tools.skills_hub._write_index_cache")
|
|
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
|
@patch("tools.skills_hub.httpx.get")
|
|
def test_fetch_downloads_skill_files_from_well_known_endpoint(self, mock_get, _mock_read_cache, _mock_write_cache):
|
|
def fake_get(url, *args, **kwargs):
|
|
if url.endswith("/index.json"):
|
|
return MagicMock(status_code=200, json=lambda: {
|
|
"skills": [{
|
|
"name": "code-review",
|
|
"description": "Review code",
|
|
"files": ["SKILL.md", "references/checklist.md"],
|
|
}]
|
|
})
|
|
if url.endswith("/code-review/SKILL.md"):
|
|
return MagicMock(status_code=200, text="# Code Review\n")
|
|
if url.endswith("/code-review/references/checklist.md"):
|
|
return MagicMock(status_code=200, text="- [ ] security\n")
|
|
raise AssertionError(url)
|
|
|
|
mock_get.side_effect = fake_get
|
|
|
|
bundle = self._source().fetch("well-known:https://example.com/.well-known/skills/code-review")
|
|
|
|
assert bundle is not None
|
|
assert bundle.source == "well-known"
|
|
assert bundle.files["SKILL.md"] == "# Code Review\n"
|
|
assert bundle.files["references/checklist.md"] == "- [ ] security\n"
|
|
|
|
|
|
class TestCheckForSkillUpdates:
|
|
def test_bundle_content_hash_matches_installed_content_hash(self, tmp_path):
|
|
from tools.skills_guard import content_hash
|
|
|
|
bundle = SkillBundle(
|
|
name="demo-skill",
|
|
files={
|
|
"SKILL.md": "same content",
|
|
"references/checklist.md": "- [ ] security\n",
|
|
},
|
|
source="github",
|
|
identifier="owner/repo/demo-skill",
|
|
trust_level="community",
|
|
)
|
|
skill_dir = tmp_path / "demo-skill"
|
|
skill_dir.mkdir()
|
|
(skill_dir / "SKILL.md").write_text("same content")
|
|
(skill_dir / "references").mkdir()
|
|
(skill_dir / "references" / "checklist.md").write_text("- [ ] security\n")
|
|
|
|
assert bundle_content_hash(bundle) == content_hash(skill_dir)
|
|
|
|
def test_reports_update_when_remote_hash_differs(self):
|
|
lock = MagicMock()
|
|
lock.list_installed.return_value = [{
|
|
"name": "demo-skill",
|
|
"source": "github",
|
|
"identifier": "owner/repo/demo-skill",
|
|
"content_hash": "oldhash",
|
|
"install_path": "demo-skill",
|
|
}]
|
|
|
|
source = MagicMock()
|
|
source.source_id.return_value = "github"
|
|
source.fetch.return_value = SkillBundle(
|
|
name="demo-skill",
|
|
files={"SKILL.md": "new content"},
|
|
source="github",
|
|
identifier="owner/repo/demo-skill",
|
|
trust_level="community",
|
|
)
|
|
|
|
results = check_for_skill_updates(lock=lock, sources=[source])
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["name"] == "demo-skill"
|
|
assert results[0]["status"] == "update_available"
|
|
|
|
def test_reports_up_to_date_when_hash_matches(self):
|
|
bundle = SkillBundle(
|
|
name="demo-skill",
|
|
files={"SKILL.md": "same content"},
|
|
source="github",
|
|
identifier="owner/repo/demo-skill",
|
|
trust_level="community",
|
|
)
|
|
lock = MagicMock()
|
|
lock.list_installed.return_value = [{
|
|
"name": "demo-skill",
|
|
"source": "github",
|
|
"identifier": "owner/repo/demo-skill",
|
|
"content_hash": bundle_content_hash(bundle),
|
|
"install_path": "demo-skill",
|
|
}]
|
|
source = MagicMock()
|
|
source.source_id.return_value = "github"
|
|
source.fetch.return_value = bundle
|
|
|
|
results = check_for_skill_updates(lock=lock, sources=[source])
|
|
|
|
assert results[0]["status"] == "up_to_date"
|
|
|
|
|
|
class TestCreateSourceRouter:
|
|
def test_includes_skills_sh_source(self):
|
|
sources = create_source_router(auth=MagicMock(spec=GitHubAuth))
|
|
assert any(isinstance(src, SkillsShSource) for src in sources)
|
|
|
|
def test_includes_well_known_source(self):
|
|
sources = create_source_router(auth=MagicMock(spec=GitHubAuth))
|
|
assert any(isinstance(src, WellKnownSkillSource) for src in sources)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HubLockFile
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHubLockFile:
|
|
def test_load_missing_file(self, tmp_path):
|
|
lock = HubLockFile(path=tmp_path / "lock.json")
|
|
data = lock.load()
|
|
assert data == {"version": 1, "installed": {}}
|
|
|
|
def test_load_valid_file(self, tmp_path):
|
|
lock_file = tmp_path / "lock.json"
|
|
lock_file.write_text(json.dumps({
|
|
"version": 1,
|
|
"installed": {"my-skill": {"source": "github"}}
|
|
}))
|
|
lock = HubLockFile(path=lock_file)
|
|
data = lock.load()
|
|
assert "my-skill" in data["installed"]
|
|
|
|
def test_load_corrupt_json(self, tmp_path):
|
|
lock_file = tmp_path / "lock.json"
|
|
lock_file.write_text("not json{{{")
|
|
lock = HubLockFile(path=lock_file)
|
|
data = lock.load()
|
|
assert data == {"version": 1, "installed": {}}
|
|
|
|
def test_save_creates_parent_dir(self, tmp_path):
|
|
lock_file = tmp_path / "subdir" / "lock.json"
|
|
lock = HubLockFile(path=lock_file)
|
|
lock.save({"version": 1, "installed": {}})
|
|
assert lock_file.exists()
|
|
|
|
def test_record_install(self, tmp_path):
|
|
lock = HubLockFile(path=tmp_path / "lock.json")
|
|
lock.record_install(
|
|
name="test-skill",
|
|
source="github",
|
|
identifier="owner/repo/test-skill",
|
|
trust_level="trusted",
|
|
scan_verdict="pass",
|
|
skill_hash="abc123",
|
|
install_path="test-skill",
|
|
files=["SKILL.md", "references/api.md"],
|
|
)
|
|
data = lock.load()
|
|
assert "test-skill" in data["installed"]
|
|
entry = data["installed"]["test-skill"]
|
|
assert entry["source"] == "github"
|
|
assert entry["trust_level"] == "trusted"
|
|
assert entry["content_hash"] == "abc123"
|
|
assert "installed_at" in entry
|
|
|
|
def test_record_uninstall(self, tmp_path):
|
|
lock = HubLockFile(path=tmp_path / "lock.json")
|
|
lock.record_install(
|
|
name="test-skill", source="github", identifier="x",
|
|
trust_level="community", scan_verdict="pass",
|
|
skill_hash="h", install_path="test-skill", files=["SKILL.md"],
|
|
)
|
|
lock.record_uninstall("test-skill")
|
|
data = lock.load()
|
|
assert "test-skill" not in data["installed"]
|
|
|
|
def test_record_uninstall_nonexistent(self, tmp_path):
|
|
lock = HubLockFile(path=tmp_path / "lock.json")
|
|
lock.save({"version": 1, "installed": {}})
|
|
# Should not raise
|
|
lock.record_uninstall("nonexistent")
|
|
|
|
def test_get_installed(self, tmp_path):
|
|
lock = HubLockFile(path=tmp_path / "lock.json")
|
|
lock.record_install(
|
|
name="skill-a", source="github", identifier="x",
|
|
trust_level="trusted", scan_verdict="pass",
|
|
skill_hash="h", install_path="skill-a", files=["SKILL.md"],
|
|
)
|
|
assert lock.get_installed("skill-a") is not None
|
|
assert lock.get_installed("nonexistent") is None
|
|
|
|
def test_list_installed(self, tmp_path):
|
|
lock = HubLockFile(path=tmp_path / "lock.json")
|
|
lock.record_install(
|
|
name="s1", source="github", identifier="x",
|
|
trust_level="trusted", scan_verdict="pass",
|
|
skill_hash="h1", install_path="s1", files=["SKILL.md"],
|
|
)
|
|
lock.record_install(
|
|
name="s2", source="clawhub", identifier="y",
|
|
trust_level="community", scan_verdict="pass",
|
|
skill_hash="h2", install_path="s2", files=["SKILL.md"],
|
|
)
|
|
installed = lock.list_installed()
|
|
assert len(installed) == 2
|
|
names = {e["name"] for e in installed}
|
|
assert names == {"s1", "s2"}
|
|
|
|
def test_is_hub_installed(self, tmp_path):
|
|
lock = HubLockFile(path=tmp_path / "lock.json")
|
|
lock.record_install(
|
|
name="my-skill", source="github", identifier="x",
|
|
trust_level="trusted", scan_verdict="pass",
|
|
skill_hash="h", install_path="my-skill", files=["SKILL.md"],
|
|
)
|
|
assert lock.is_hub_installed("my-skill") is True
|
|
assert lock.is_hub_installed("other") is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TapsManager
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTapsManager:
|
|
def test_load_missing_file(self, tmp_path):
|
|
mgr = TapsManager(path=tmp_path / "taps.json")
|
|
assert mgr.load() == []
|
|
|
|
def test_load_valid_file(self, tmp_path):
|
|
taps_file = tmp_path / "taps.json"
|
|
taps_file.write_text(json.dumps({"taps": [{"repo": "owner/repo", "path": "skills/"}]}))
|
|
mgr = TapsManager(path=taps_file)
|
|
taps = mgr.load()
|
|
assert len(taps) == 1
|
|
assert taps[0]["repo"] == "owner/repo"
|
|
|
|
def test_load_corrupt_json(self, tmp_path):
|
|
taps_file = tmp_path / "taps.json"
|
|
taps_file.write_text("bad json")
|
|
mgr = TapsManager(path=taps_file)
|
|
assert mgr.load() == []
|
|
|
|
def test_add_new_tap(self, tmp_path):
|
|
mgr = TapsManager(path=tmp_path / "taps.json")
|
|
assert mgr.add("owner/repo", "skills/") is True
|
|
taps = mgr.load()
|
|
assert len(taps) == 1
|
|
assert taps[0]["repo"] == "owner/repo"
|
|
|
|
def test_add_duplicate_tap(self, tmp_path):
|
|
mgr = TapsManager(path=tmp_path / "taps.json")
|
|
mgr.add("owner/repo")
|
|
assert mgr.add("owner/repo") is False
|
|
assert len(mgr.load()) == 1
|
|
|
|
def test_remove_existing_tap(self, tmp_path):
|
|
mgr = TapsManager(path=tmp_path / "taps.json")
|
|
mgr.add("owner/repo")
|
|
assert mgr.remove("owner/repo") is True
|
|
assert mgr.load() == []
|
|
|
|
def test_remove_nonexistent_tap(self, tmp_path):
|
|
mgr = TapsManager(path=tmp_path / "taps.json")
|
|
assert mgr.remove("nonexistent") is False
|
|
|
|
def test_list_taps(self, tmp_path):
|
|
mgr = TapsManager(path=tmp_path / "taps.json")
|
|
mgr.add("repo-a/skills")
|
|
mgr.add("repo-b/tools")
|
|
taps = mgr.list_taps()
|
|
assert len(taps) == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LobeHubSource._convert_to_skill_md
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestConvertToSkillMd:
|
|
def test_basic_conversion(self):
|
|
agent_data = {
|
|
"identifier": "test-agent",
|
|
"meta": {
|
|
"title": "Test Agent",
|
|
"description": "A test agent.",
|
|
"tags": ["testing", "demo"],
|
|
},
|
|
"config": {
|
|
"systemRole": "You are a helpful test agent.",
|
|
},
|
|
}
|
|
result = LobeHubSource._convert_to_skill_md(agent_data)
|
|
assert "---" in result
|
|
assert "name: test-agent" in result
|
|
assert "description: A test agent." in result
|
|
assert "tags: [testing, demo]" in result
|
|
assert "# Test Agent" in result
|
|
assert "You are a helpful test agent." in result
|
|
|
|
def test_missing_system_role(self):
|
|
agent_data = {
|
|
"identifier": "no-role",
|
|
"meta": {"title": "No Role", "description": "Desc."},
|
|
}
|
|
result = LobeHubSource._convert_to_skill_md(agent_data)
|
|
assert "(No system role defined)" in result
|
|
|
|
def test_missing_meta(self):
|
|
agent_data = {"identifier": "bare-agent"}
|
|
result = LobeHubSource._convert_to_skill_md(agent_data)
|
|
assert "name: bare-agent" in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# unified_search — dedup logic
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUnifiedSearchDedup:
|
|
def _make_source(self, source_id, results):
|
|
"""Create a mock SkillSource that returns fixed results."""
|
|
src = MagicMock()
|
|
src.source_id.return_value = source_id
|
|
src.search.return_value = results
|
|
return src
|
|
|
|
def test_dedup_keeps_first_seen(self):
|
|
s1 = SkillMeta(name="skill", description="from A", source="a",
|
|
identifier="a/skill", trust_level="community")
|
|
s2 = SkillMeta(name="skill", description="from B", source="b",
|
|
identifier="b/skill", trust_level="community")
|
|
src_a = self._make_source("a", [s1])
|
|
src_b = self._make_source("b", [s2])
|
|
results = unified_search("skill", [src_a, src_b])
|
|
assert len(results) == 1
|
|
assert results[0].description == "from A"
|
|
|
|
def test_dedup_prefers_trusted_over_community(self):
|
|
community = SkillMeta(name="skill", description="community", source="a",
|
|
identifier="a/skill", trust_level="community")
|
|
trusted = SkillMeta(name="skill", description="trusted", source="b",
|
|
identifier="b/skill", trust_level="trusted")
|
|
src_a = self._make_source("a", [community])
|
|
src_b = self._make_source("b", [trusted])
|
|
results = unified_search("skill", [src_a, src_b])
|
|
assert len(results) == 1
|
|
assert results[0].trust_level == "trusted"
|
|
|
|
def test_dedup_prefers_builtin_over_trusted(self):
|
|
"""Regression: builtin must not be overwritten by trusted."""
|
|
builtin = SkillMeta(name="skill", description="builtin", source="a",
|
|
identifier="a/skill", trust_level="builtin")
|
|
trusted = SkillMeta(name="skill", description="trusted", source="b",
|
|
identifier="b/skill", trust_level="trusted")
|
|
src_a = self._make_source("a", [builtin])
|
|
src_b = self._make_source("b", [trusted])
|
|
results = unified_search("skill", [src_a, src_b])
|
|
assert len(results) == 1
|
|
assert results[0].trust_level == "builtin"
|
|
|
|
def test_dedup_trusted_not_overwritten_by_community(self):
|
|
trusted = SkillMeta(name="skill", description="trusted", source="a",
|
|
identifier="a/skill", trust_level="trusted")
|
|
community = SkillMeta(name="skill", description="community", source="b",
|
|
identifier="b/skill", trust_level="community")
|
|
src_a = self._make_source("a", [trusted])
|
|
src_b = self._make_source("b", [community])
|
|
results = unified_search("skill", [src_a, src_b])
|
|
assert results[0].trust_level == "trusted"
|
|
|
|
def test_source_filter(self):
|
|
s1 = SkillMeta(name="s1", description="d", source="a",
|
|
identifier="x", trust_level="community")
|
|
s2 = SkillMeta(name="s2", description="d", source="b",
|
|
identifier="y", trust_level="community")
|
|
src_a = self._make_source("a", [s1])
|
|
src_b = self._make_source("b", [s2])
|
|
results = unified_search("query", [src_a, src_b], source_filter="a")
|
|
assert len(results) == 1
|
|
assert results[0].name == "s1"
|
|
|
|
def test_limit_respected(self):
|
|
skills = [
|
|
SkillMeta(name=f"s{i}", description="d", source="a",
|
|
identifier=f"a/s{i}", trust_level="community")
|
|
for i in range(20)
|
|
]
|
|
src = self._make_source("a", skills)
|
|
results = unified_search("query", [src], limit=5)
|
|
assert len(results) == 5
|
|
|
|
def test_source_error_handled(self):
|
|
failing = MagicMock()
|
|
failing.source_id.return_value = "fail"
|
|
failing.search.side_effect = RuntimeError("boom")
|
|
ok = self._make_source("ok", [
|
|
SkillMeta(name="s1", description="d", source="ok",
|
|
identifier="x", trust_level="community")
|
|
])
|
|
results = unified_search("query", [failing, ok])
|
|
assert len(results) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# append_audit_log
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAppendAuditLog:
|
|
def test_creates_log_entry(self, tmp_path):
|
|
log_file = tmp_path / "audit.log"
|
|
with patch("tools.skills_hub.AUDIT_LOG", log_file):
|
|
append_audit_log("INSTALL", "test-skill", "github", "trusted", "pass")
|
|
content = log_file.read_text()
|
|
assert "INSTALL" in content
|
|
assert "test-skill" in content
|
|
assert "github:trusted" in content
|
|
assert "pass" in content
|
|
|
|
def test_appends_multiple_entries(self, tmp_path):
|
|
log_file = tmp_path / "audit.log"
|
|
with patch("tools.skills_hub.AUDIT_LOG", log_file):
|
|
append_audit_log("INSTALL", "s1", "github", "trusted", "pass")
|
|
append_audit_log("UNINSTALL", "s1", "github", "trusted", "n/a")
|
|
lines = log_file.read_text().strip().split("\n")
|
|
assert len(lines) == 2
|
|
|
|
def test_extra_field_included(self, tmp_path):
|
|
log_file = tmp_path / "audit.log"
|
|
with patch("tools.skills_hub.AUDIT_LOG", log_file):
|
|
append_audit_log("INSTALL", "s1", "github", "trusted", "pass", extra="hash123")
|
|
content = log_file.read_text()
|
|
assert "hash123" in content
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _skill_meta_to_dict
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSkillMetaToDict:
|
|
def test_roundtrip(self):
|
|
meta = SkillMeta(
|
|
name="test", description="desc", source="github",
|
|
identifier="owner/repo/test", trust_level="trusted",
|
|
repo="owner/repo", path="skills/test", tags=["a", "b"],
|
|
)
|
|
d = _skill_meta_to_dict(meta)
|
|
assert d["name"] == "test"
|
|
assert d["tags"] == ["a", "b"]
|
|
# Can reconstruct from dict
|
|
restored = SkillMeta(**d)
|
|
assert restored.name == meta.name
|
|
assert restored.trust_level == meta.trust_level
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Official skills / binary assets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOptionalSkillSourceBinaryAssets:
|
|
def test_fetch_preserves_binary_assets(self, tmp_path):
|
|
optional_root = tmp_path / "optional-skills"
|
|
skill_dir = optional_root / "mlops" / "models" / "neutts"
|
|
(skill_dir / "assets" / "neutts-cli" / "samples").mkdir(parents=True)
|
|
(skill_dir / "SKILL.md").write_text(
|
|
"---\nname: neutts\ndescription: test\n---\n\nBody\n",
|
|
encoding="utf-8",
|
|
)
|
|
wav_bytes = b"RIFF\x00\x01fakewav"
|
|
(skill_dir / "assets" / "neutts-cli" / "samples" / "jo.wav").write_bytes(
|
|
wav_bytes
|
|
)
|
|
(skill_dir / "assets" / "neutts-cli" / "samples" / "jo.txt").write_text(
|
|
"hello\n", encoding="utf-8"
|
|
)
|
|
pycache_dir = skill_dir / "assets" / "neutts-cli" / "src" / "neutts_cli" / "__pycache__"
|
|
pycache_dir.mkdir(parents=True)
|
|
(pycache_dir / "cli.cpython-312.pyc").write_bytes(b"junk")
|
|
|
|
src = OptionalSkillSource()
|
|
src._optional_dir = optional_root
|
|
|
|
bundle = src.fetch("official/mlops/models/neutts")
|
|
|
|
assert bundle is not None
|
|
assert bundle.files["assets/neutts-cli/samples/jo.wav"] == wav_bytes
|
|
assert bundle.files["assets/neutts-cli/samples/jo.txt"] == b"hello\n"
|
|
assert "assets/neutts-cli/src/neutts_cli/__pycache__/cli.cpython-312.pyc" not in bundle.files
|
|
|
|
|
|
class TestQuarantineBundleBinaryAssets:
|
|
def test_quarantine_bundle_writes_binary_files(self, tmp_path):
|
|
import tools.skills_hub as hub
|
|
|
|
hub_dir = tmp_path / "skills" / ".hub"
|
|
with patch.object(hub, "SKILLS_DIR", tmp_path / "skills"), \
|
|
patch.object(hub, "HUB_DIR", hub_dir), \
|
|
patch.object(hub, "LOCK_FILE", hub_dir / "lock.json"), \
|
|
patch.object(hub, "QUARANTINE_DIR", hub_dir / "quarantine"), \
|
|
patch.object(hub, "AUDIT_LOG", hub_dir / "audit.log"), \
|
|
patch.object(hub, "TAPS_FILE", hub_dir / "taps.json"), \
|
|
patch.object(hub, "INDEX_CACHE_DIR", hub_dir / "index-cache"):
|
|
bundle = SkillBundle(
|
|
name="neutts",
|
|
files={
|
|
"SKILL.md": "---\nname: neutts\n---\n",
|
|
"assets/neutts-cli/samples/jo.wav": b"RIFF\x00\x01fakewav",
|
|
},
|
|
source="official",
|
|
identifier="official/mlops/models/neutts",
|
|
trust_level="builtin",
|
|
)
|
|
|
|
q_path = quarantine_bundle(bundle)
|
|
|
|
assert (q_path / "SKILL.md").read_text(encoding="utf-8").startswith("---")
|
|
assert (q_path / "assets" / "neutts-cli" / "samples" / "jo.wav").read_bytes() == b"RIFF\x00\x01fakewav"
|