Compare commits
2 Commits
fix/issue-
...
fix/754
| Author | SHA1 | Date | |
|---|---|---|---|
| 244dae106d | |||
| a606660c9d |
81
tests/test_skill_deps.py
Normal file
81
tests/test_skill_deps.py
Normal file
@@ -0,0 +1,81 @@
|
||||
"""
|
||||
Tests for skill dependency resolver
|
||||
|
||||
Issue: #754
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from tools.skill_deps import (
|
||||
parse_requires,
|
||||
check_dependency,
|
||||
check_dependencies,
|
||||
resolve_dependencies,
|
||||
)
|
||||
|
||||
|
||||
class TestParseRequires(unittest.TestCase):
|
||||
|
||||
def test_list(self):
|
||||
fm = {"requires": ["pkg1", "pkg2"]}
|
||||
self.assertEqual(parse_requires(fm), ["pkg1", "pkg2"])
|
||||
|
||||
def test_string(self):
|
||||
fm = {"requires": "pkg1"}
|
||||
self.assertEqual(parse_requires(fm), ["pkg1"])
|
||||
|
||||
def test_empty(self):
|
||||
fm = {}
|
||||
self.assertEqual(parse_requires(fm), [])
|
||||
|
||||
def test_none(self):
|
||||
fm = {"requires": None}
|
||||
self.assertEqual(parse_requires(fm), [])
|
||||
|
||||
|
||||
class TestCheckDependency(unittest.TestCase):
|
||||
|
||||
def test_installed(self):
|
||||
installed, version = check_dependency("json")
|
||||
self.assertTrue(installed)
|
||||
|
||||
def test_not_installed(self):
|
||||
installed, version = check_dependency("nonexistent_package_xyz_123")
|
||||
self.assertFalse(installed)
|
||||
|
||||
|
||||
class TestCheckDependencies(unittest.TestCase):
|
||||
|
||||
def test_all_installed(self):
|
||||
installed, missing = check_dependencies(["json", "os"])
|
||||
self.assertEqual(len(missing), 0)
|
||||
|
||||
def test_some_missing(self):
|
||||
installed, missing = check_dependencies(["json", "nonexistent_xyz"])
|
||||
self.assertIn("json", installed)
|
||||
self.assertIn("nonexistent_xyz", missing)
|
||||
|
||||
|
||||
class TestResolveDependencies(unittest.TestCase):
|
||||
|
||||
def test_no_requires(self):
|
||||
satisfied, installed, missing = resolve_dependencies({})
|
||||
self.assertTrue(satisfied)
|
||||
|
||||
def test_all_satisfied(self):
|
||||
satisfied, installed, missing = resolve_dependencies(
|
||||
{"requires": ["json"]}, auto_install=False
|
||||
)
|
||||
self.assertTrue(satisfied)
|
||||
|
||||
def test_missing_no_auto(self):
|
||||
satisfied, installed, missing = resolve_dependencies(
|
||||
{"requires": ["nonexistent_xyz"]}, auto_install=False
|
||||
)
|
||||
self.assertFalse(satisfied)
|
||||
self.assertIn("nonexistent_xyz", missing)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
185
tools/skill_deps.py
Normal file
185
tools/skill_deps.py
Normal file
@@ -0,0 +1,185 @@
|
||||
"""
|
||||
Skill Dependency Resolver — Auto-install missing dependencies
|
||||
|
||||
Checks skill frontmatter for `requires` field and ensures
|
||||
dependencies are installed before loading the skill.
|
||||
|
||||
Issue: #754
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import logging
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def parse_requires(frontmatter: Dict[str, Any]) -> List[str]:
|
||||
"""
|
||||
Parse the `requires` field from skill frontmatter.
|
||||
|
||||
Supports:
|
||||
- requires: [package1, package2]
|
||||
- requires: package1
|
||||
- requires:
|
||||
- package1
|
||||
- package2
|
||||
"""
|
||||
requires = frontmatter.get("requires", [])
|
||||
if isinstance(requires, str):
|
||||
return [requires]
|
||||
if isinstance(requires, list):
|
||||
return [str(r) for r in requires if r]
|
||||
return []
|
||||
|
||||
|
||||
def check_dependency(package: str) -> Tuple[bool, str]:
|
||||
"""
|
||||
Check if a Python package is installed.
|
||||
|
||||
Returns:
|
||||
Tuple of (is_installed, version_or_error)
|
||||
"""
|
||||
# Handle pip package names (e.g., "matrix-nio[e2e]")
|
||||
import_name = package.split("[")[0].replace("-", "_")
|
||||
|
||||
try:
|
||||
mod = importlib.import_module(import_name)
|
||||
version = getattr(mod, "__version__", "installed")
|
||||
return True, version
|
||||
except ImportError:
|
||||
return False, "not installed"
|
||||
|
||||
|
||||
def check_dependencies(requires: List[str]) -> Tuple[List[str], List[str]]:
|
||||
"""
|
||||
Check which dependencies are missing.
|
||||
|
||||
Returns:
|
||||
Tuple of (installed, missing)
|
||||
"""
|
||||
installed = []
|
||||
missing = []
|
||||
|
||||
for pkg in requires:
|
||||
is_installed, _ = check_dependency(pkg)
|
||||
if is_installed:
|
||||
installed.append(pkg)
|
||||
else:
|
||||
missing.append(pkg)
|
||||
|
||||
return installed, missing
|
||||
|
||||
|
||||
def install_dependency(package: str, quiet: bool = False) -> Tuple[bool, str]:
|
||||
"""
|
||||
Install a Python package via pip.
|
||||
|
||||
Returns:
|
||||
Tuple of (success, output_or_error)
|
||||
"""
|
||||
try:
|
||||
cmd = [sys.executable, "-m", "pip", "install", package]
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
if not quiet:
|
||||
logger.info("Installed %s", package)
|
||||
return True, result.stdout
|
||||
else:
|
||||
logger.error("Failed to install %s: %s", package, result.stderr)
|
||||
return False, result.stderr
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return False, "Installation timed out"
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
|
||||
|
||||
def resolve_dependencies(
|
||||
frontmatter: Dict[str, Any],
|
||||
auto_install: bool = False,
|
||||
quiet: bool = False
|
||||
) -> Tuple[bool, List[str], List[str]]:
|
||||
"""
|
||||
Resolve skill dependencies.
|
||||
|
||||
Args:
|
||||
frontmatter: Skill frontmatter dict
|
||||
auto_install: If True, install missing deps without asking
|
||||
quiet: If True, suppress output
|
||||
|
||||
Returns:
|
||||
Tuple of (all_satisfied, installed_now, still_missing)
|
||||
"""
|
||||
requires = parse_requires(frontmatter)
|
||||
if not requires:
|
||||
return True, [], []
|
||||
|
||||
installed, missing = check_dependencies(requires)
|
||||
|
||||
if not missing:
|
||||
if not quiet:
|
||||
logger.debug("All dependencies satisfied: %s", installed)
|
||||
return True, [], []
|
||||
|
||||
if not auto_install:
|
||||
if not quiet:
|
||||
logger.warning("Missing dependencies: %s", missing)
|
||||
return False, [], missing
|
||||
|
||||
# Auto-install missing dependencies
|
||||
installed_now = []
|
||||
still_missing = []
|
||||
|
||||
for pkg in missing:
|
||||
if not quiet:
|
||||
logger.info("Installing missing dependency: %s", pkg)
|
||||
|
||||
success, output = install_dependency(pkg, quiet=quiet)
|
||||
if success:
|
||||
installed_now.append(pkg)
|
||||
else:
|
||||
still_missing.append(pkg)
|
||||
logger.error("Failed to install %s: %s", pkg, output[:200])
|
||||
|
||||
all_satisfied = len(still_missing) == 0
|
||||
return all_satisfied, installed_now, still_missing
|
||||
|
||||
|
||||
def check_skill_dependencies(skill_dir) -> Dict[str, Any]:
|
||||
"""
|
||||
Check dependencies for a skill directory.
|
||||
|
||||
Returns:
|
||||
Dict with dependency status
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
skill_md = Path(skill_dir) / "SKILL.md"
|
||||
if not skill_md.exists():
|
||||
return {"requires": [], "installed": [], "missing": [], "satisfied": True}
|
||||
|
||||
try:
|
||||
content = skill_md.read_text()
|
||||
from agent.skill_utils import parse_frontmatter
|
||||
frontmatter, _ = parse_frontmatter(content)
|
||||
except Exception:
|
||||
return {"requires": [], "installed": [], "missing": [], "satisfied": True}
|
||||
|
||||
requires = parse_requires(frontmatter)
|
||||
installed, missing = check_dependencies(requires)
|
||||
|
||||
return {
|
||||
"requires": requires,
|
||||
"installed": installed,
|
||||
"missing": missing,
|
||||
"satisfied": len(missing) == 0
|
||||
}
|
||||
Reference in New Issue
Block a user