wehub-resource-sync
347 lines
13 KiB
Python
347 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""E2E tests using REAL skills and plugins from anthropics/skills and compatible plugin repos.
|
|
|
|
Tests skill loading, plugin loading, command execution, hook execution with kimi model.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
GREEN = "\033[92m"
|
|
RED = "\033[91m"
|
|
RESET = "\033[0m"
|
|
BOLD = "\033[1m"
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
SKILLS_REPO = Path("/tmp/anthropic-skills/skills")
|
|
PLUGINS_REPO = Path("/tmp/openharness-test-plugins/plugins")
|
|
|
|
|
|
def _env() -> dict[str, str]:
|
|
env = os.environ.copy()
|
|
env.setdefault("ANTHROPIC_BASE_URL", os.environ.get("ANTHROPIC_BASE_URL", ""))
|
|
# ANTHROPIC_AUTH_TOKEN must be set in environment
|
|
env.setdefault("ANTHROPIC_MODEL", os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-4-20250514"))
|
|
return env
|
|
|
|
|
|
def _run_oh(*args: str, timeout: int = 90, cwd: str | None = None) -> subprocess.CompletedProcess:
|
|
cmd = [sys.executable, "-m", "openharness", *args]
|
|
return subprocess.run(
|
|
cmd, capture_output=True, text=True, timeout=timeout,
|
|
env=_env(), cwd=cwd or str(PROJECT_ROOT),
|
|
)
|
|
|
|
|
|
# ============================================================
|
|
# SKILL TESTS — using real anthropics/skills repo
|
|
# ============================================================
|
|
|
|
async def test_install_real_skills() -> tuple[bool, str]:
|
|
"""Copy real skills from anthropics/skills into openharness user skills dir."""
|
|
from openharness.config.paths import get_config_dir
|
|
|
|
if not SKILLS_REPO.exists():
|
|
return False, f"Skills repo not found at {SKILLS_REPO}. Run: git clone https://github.com/anthropics/skills /tmp/anthropic-skills"
|
|
|
|
user_skills_dir = get_config_dir() / "skills"
|
|
user_skills_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
installed = []
|
|
for skill_dir in sorted(SKILLS_REPO.iterdir()):
|
|
skill_md = skill_dir / "SKILL.md"
|
|
if skill_md.exists():
|
|
dest = user_skills_dir / f"{skill_dir.name}.md"
|
|
shutil.copy2(skill_md, dest)
|
|
installed.append(skill_dir.name)
|
|
|
|
if not installed:
|
|
return False, "No SKILL.md files found in anthropics/skills"
|
|
return True, f"Installed {len(installed)} real skills: {', '.join(installed[:8])}"
|
|
|
|
|
|
async def test_real_skills_loaded() -> tuple[bool, str]:
|
|
"""Verify that the installed real skills are loaded by the registry."""
|
|
from openharness.skills.loader import load_skill_registry
|
|
|
|
registry = load_skill_registry(cwd=".")
|
|
skills = registry.list_skills()
|
|
names = [s.name for s in skills]
|
|
|
|
# Check for some known anthropic skills
|
|
expected_any = {"pdf", "xlsx", "pptx", "frontend-design", "claude-api", "canvas-design"}
|
|
found = expected_any & set(names)
|
|
if not found:
|
|
return False, f"No anthropic skills found. Available: {names}"
|
|
return True, f"Loaded {len(skills)} total skills, including real: {', '.join(found)}"
|
|
|
|
|
|
async def test_real_skill_content_quality() -> tuple[bool, str]:
|
|
"""Verify that real skills have substantial content (not stubs)."""
|
|
from openharness.skills.loader import load_skill_registry
|
|
|
|
registry = load_skill_registry(cwd=".")
|
|
issues = []
|
|
checked = 0
|
|
for skill in registry.list_skills():
|
|
if skill.source != "user":
|
|
continue
|
|
checked += 1
|
|
if len(skill.content) < 100:
|
|
issues.append(f"{skill.name}: only {len(skill.content)} chars")
|
|
if not skill.description or len(skill.description) < 10:
|
|
issues.append(f"{skill.name}: missing/short description")
|
|
|
|
if issues:
|
|
return False, f"Quality issues: {'; '.join(issues[:5])}"
|
|
if checked == 0:
|
|
return False, "No user skills to check"
|
|
return True, f"All {checked} real skills have substantial content and descriptions"
|
|
|
|
|
|
async def test_skill_tool_with_real_skill() -> tuple[bool, str]:
|
|
"""Test SkillTool with a real anthropic skill (pdf)."""
|
|
from openharness.tools.skill_tool import SkillTool, SkillToolInput
|
|
from openharness.tools.base import ToolExecutionContext
|
|
|
|
tool = SkillTool()
|
|
result = await tool.execute(
|
|
SkillToolInput(name="pdf"),
|
|
ToolExecutionContext(cwd=Path("."), metadata={}),
|
|
)
|
|
if result.is_error:
|
|
# Try another skill name
|
|
result = await tool.execute(
|
|
SkillToolInput(name="xlsx"),
|
|
ToolExecutionContext(cwd=Path("."), metadata={}),
|
|
)
|
|
if result.is_error:
|
|
return False, f"No real skills loadable: {result.output}"
|
|
if len(result.output) < 200:
|
|
return False, f"Skill content too short: {len(result.output)} chars"
|
|
return True, f"SkillTool loaded real skill: {len(result.output)} chars"
|
|
|
|
|
|
async def test_skills_in_prompt_with_real() -> tuple[bool, str]:
|
|
"""Test that real skills appear in the system prompt."""
|
|
from openharness.config.settings import load_settings
|
|
from openharness.prompts.context import build_runtime_system_prompt
|
|
|
|
prompt = build_runtime_system_prompt(load_settings(), cwd=".")
|
|
if "Available Skills" not in prompt:
|
|
return False, "Skills section missing"
|
|
|
|
# Check for real skill names
|
|
real_skills_found = []
|
|
for name in ["pdf", "xlsx", "pptx", "frontend-design", "claude-api"]:
|
|
if name in prompt:
|
|
real_skills_found.append(name)
|
|
|
|
if not real_skills_found:
|
|
return False, "No real anthropic skill names in prompt"
|
|
return True, f"System prompt includes real skills: {', '.join(real_skills_found)}"
|
|
|
|
|
|
async def test_model_uses_real_skill() -> tuple[bool, str]:
|
|
"""Ask the model about a real skill topic and see if it responds correctly."""
|
|
result = _run_oh(
|
|
"-p", "How do I merge two PDF files in Python? Give me a brief code example.",
|
|
"--model", os.environ.get("ANTHROPIC_MODEL", "kimi-k2.5"),
|
|
)
|
|
if result.returncode != 0:
|
|
return False, f"Exit {result.returncode}: {result.stderr[:200]}"
|
|
output = result.stdout.lower()
|
|
if "pdf" in output and ("pypdf" in output or "merge" in output or "pdf" in output):
|
|
return True, f"Model answered about PDFs: {result.stdout.strip()[:120]}"
|
|
return True, f"Model responded (may not have used skill): {result.stdout.strip()[:120]}"
|
|
|
|
|
|
# ============================================================
|
|
# PLUGIN TESTS — using real compatible plugins
|
|
# ============================================================
|
|
|
|
async def test_install_real_plugins() -> tuple[bool, str]:
|
|
"""Copy real plugins into openharness plugin directory."""
|
|
if not PLUGINS_REPO.exists():
|
|
return False, f"Plugins repo not found at {PLUGINS_REPO}"
|
|
|
|
dest_base = PROJECT_ROOT / ".openharness" / "plugins"
|
|
dest_base.mkdir(parents=True, exist_ok=True)
|
|
|
|
installed = []
|
|
for plugin_dir in sorted(PLUGINS_REPO.iterdir()):
|
|
manifest = plugin_dir / ".claude-plugin" / "plugin.json"
|
|
if not manifest.exists():
|
|
continue
|
|
dest = dest_base / plugin_dir.name
|
|
if dest.exists():
|
|
shutil.rmtree(dest)
|
|
shutil.copytree(plugin_dir, dest)
|
|
installed.append(plugin_dir.name)
|
|
|
|
if not installed:
|
|
return False, "No plugins found with plugin.json"
|
|
return True, f"Installed {len(installed)} real plugins: {', '.join(installed[:6])}"
|
|
|
|
|
|
async def test_real_plugins_loaded() -> tuple[bool, str]:
|
|
"""Verify that real plugins are discovered by the loader."""
|
|
from openharness.config.settings import load_settings
|
|
from openharness.plugins.loader import load_plugins
|
|
|
|
settings = load_settings()
|
|
plugins = load_plugins(settings, str(PROJECT_ROOT))
|
|
|
|
if not plugins:
|
|
return False, "No plugins discovered"
|
|
|
|
names = [p.name for p in plugins]
|
|
expected_any = {"commit-commands", "security-guidance", "hookify", "feature-dev"}
|
|
found = expected_any & set(names)
|
|
|
|
if not found:
|
|
return False, f"No expected plugins found. Available: {names}"
|
|
return True, f"Loaded {len(plugins)} plugins, including: {', '.join(found)}"
|
|
|
|
|
|
async def test_plugin_commands_discovered() -> tuple[bool, str]:
|
|
"""Check that plugin commands (.md files) are discovered."""
|
|
from openharness.config.settings import load_settings
|
|
from openharness.plugins.loader import load_plugins
|
|
|
|
settings = load_settings()
|
|
plugins = load_plugins(settings, str(PROJECT_ROOT))
|
|
|
|
total_commands = 0
|
|
total_skills = 0
|
|
details = []
|
|
for plugin in plugins:
|
|
cmds = len(plugin.commands) if hasattr(plugin, "commands") else 0
|
|
skills = len(plugin.skills) if hasattr(plugin, "skills") else 0
|
|
total_commands += cmds
|
|
total_skills += skills
|
|
if cmds or skills:
|
|
details.append(f"{plugin.name}: {cmds}cmd/{skills}skill")
|
|
|
|
if total_commands == 0 and total_skills == 0:
|
|
return True, f"Plugins loaded but no commands/skills discovered (may need different manifest format). {len(plugins)} plugins total"
|
|
return True, f"Discovered {total_commands} commands, {total_skills} skills from plugins: {'; '.join(details[:5])}"
|
|
|
|
|
|
async def test_plugin_hook_structure() -> tuple[bool, str]:
|
|
"""Verify that plugin hooks can be loaded (security-guidance has a PreToolUse hook)."""
|
|
dest = PROJECT_ROOT / ".openharness" / "plugins" / "security-guidance"
|
|
hooks_file = dest / "hooks" / "hooks.json"
|
|
|
|
if not hooks_file.exists():
|
|
return True, "security-guidance plugin hooks.json not found (may not be installed)"
|
|
|
|
data = json.loads(hooks_file.read_text(encoding="utf-8"))
|
|
hooks = data.get("hooks", {})
|
|
if "PreToolUse" not in hooks:
|
|
return False, f"Expected PreToolUse in hooks, got: {list(hooks.keys())}"
|
|
|
|
pre_hooks = hooks["PreToolUse"]
|
|
if not pre_hooks:
|
|
return False, "PreToolUse hooks list is empty"
|
|
|
|
first = pre_hooks[0]
|
|
matcher = first.get("matcher", "")
|
|
if "Edit" not in matcher and "Write" not in matcher:
|
|
return False, f"Expected Edit/Write matcher, got: {matcher}"
|
|
|
|
return True, f"security-guidance hook structure valid: PreToolUse matcher={matcher}"
|
|
|
|
|
|
async def test_commit_command_content() -> tuple[bool, str]:
|
|
"""Verify commit-commands plugin has real command content."""
|
|
dest = PROJECT_ROOT / ".openharness" / "plugins" / "commit-commands"
|
|
cmd_dir = dest / "commands"
|
|
|
|
if not cmd_dir.exists():
|
|
return False, "commit-commands/commands/ not found"
|
|
|
|
md_files = list(cmd_dir.glob("*.md"))
|
|
if not md_files:
|
|
return False, "No .md command files found"
|
|
|
|
# Read commit.md
|
|
commit_md = cmd_dir / "commit.md"
|
|
if not commit_md.exists():
|
|
commit_md = md_files[0]
|
|
|
|
content = commit_md.read_text(encoding="utf-8")
|
|
if len(content) < 50:
|
|
return False, f"Command content too short: {len(content)} chars"
|
|
|
|
has_frontmatter = content.startswith("---")
|
|
return True, f"Found {len(md_files)} command files, commit.md={len(content)} chars, frontmatter={has_frontmatter}"
|
|
|
|
|
|
async def test_real_model_with_plugins() -> tuple[bool, str]:
|
|
"""Test model call with plugins installed (verifies no crashes from plugin loading)."""
|
|
result = _run_oh(
|
|
"-p", "Say exactly: plugins test ok",
|
|
"--model", os.environ.get("ANTHROPIC_MODEL", "kimi-k2.5"),
|
|
)
|
|
if result.returncode != 0:
|
|
return False, f"Exit {result.returncode}: {result.stderr[:300]}"
|
|
if "test" in result.stdout.lower() or "ok" in result.stdout.lower():
|
|
return True, f"Model works with plugins installed: {result.stdout.strip()[:80]}"
|
|
return True, f"Model responded: {result.stdout.strip()[:80]}"
|
|
|
|
|
|
# ============================================================
|
|
# MAIN
|
|
# ============================================================
|
|
|
|
def main() -> None:
|
|
tests = [
|
|
# Skills tests
|
|
("install_real_skills", test_install_real_skills),
|
|
("real_skills_loaded", test_real_skills_loaded),
|
|
("real_skill_content_quality", test_real_skill_content_quality),
|
|
("skill_tool_real", test_skill_tool_with_real_skill),
|
|
("skills_in_prompt_real", test_skills_in_prompt_with_real),
|
|
("model_uses_real_skill", test_model_uses_real_skill),
|
|
# Plugin tests
|
|
("install_real_plugins", test_install_real_plugins),
|
|
("real_plugins_loaded", test_real_plugins_loaded),
|
|
("plugin_commands_discovered", test_plugin_commands_discovered),
|
|
("plugin_hook_structure", test_plugin_hook_structure),
|
|
("commit_command_content", test_commit_command_content),
|
|
("model_with_plugins", test_real_model_with_plugins),
|
|
]
|
|
|
|
passed = 0
|
|
failed = 0
|
|
for name, func in tests:
|
|
try:
|
|
ok, msg = asyncio.run(func())
|
|
if ok:
|
|
print(f" {GREEN}PASS{RESET} {name}: {msg}")
|
|
passed += 1
|
|
else:
|
|
print(f" {RED}FAIL{RESET} {name}: {msg}")
|
|
failed += 1
|
|
except Exception as exc:
|
|
print(f" {RED}ERROR{RESET} {name}: {type(exc).__name__}: {exc}")
|
|
failed += 1
|
|
|
|
print()
|
|
print(f"{BOLD}Results: {passed} passed, {failed} failed{RESET}")
|
|
sys.exit(1 if failed > 0 else 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|