Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 100 additions & 0 deletions src/qwenpaw/cli/skills_cmd.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,10 @@
from __future__ import annotations

from pathlib import Path
from typing import Any

import click
import yaml

from ..agents.skills_manager import (
SkillPoolService,
Expand All @@ -14,8 +16,10 @@
reconcile_pool_manifest,
reconcile_workspace_manifest,
)
from ..agents.utils.file_handling import read_text_file_with_encoding_fallback
from ..constant import WORKING_DIR
from ..config import load_config
from ..security.skill_scanner import SkillScanError, scan_skill_directory
from .utils import prompt_checkbox, prompt_confirm


Expand Down Expand Up @@ -62,6 +66,87 @@ def _print_skill_changes(
)


def _parse_skill_frontmatter(skill_md: Path) -> dict[str, Any]:
"""Return YAML frontmatter from a skill markdown file."""
content = read_text_file_with_encoding_fallback(skill_md)
lines = content.splitlines()
if not lines or lines[0].strip() != "---":
raise click.ClickException(
"SKILL.md must start with YAML frontmatter.",
)

end_index = None
for index, line in enumerate(lines[1:], start=1):
if line.strip() in {"---", "..."}:
end_index = index
break
if end_index is None:
raise click.ClickException(
"SKILL.md frontmatter is missing a closing delimiter.",
)

try:
metadata = yaml.safe_load("\n".join(lines[1:end_index])) or {}
except yaml.YAMLError as exc:
raise click.ClickException(
f"SKILL.md frontmatter is invalid YAML: {exc}",
) from exc

if not isinstance(metadata, dict):
raise click.ClickException("SKILL.md frontmatter must be a mapping.")
return metadata


def _validate_skill_frontmatter(skill_dir: Path) -> str:
"""Validate required skill metadata and return the declared name."""
skill_md = skill_dir / "SKILL.md"
if not skill_md.is_file():
raise click.ClickException(f"Missing SKILL.md: {skill_md}")

metadata = _parse_skill_frontmatter(skill_md)
name = str(metadata.get("name") or "").strip()
description = str(metadata.get("description") or "").strip()
if not name or not description:
raise click.ClickException(
"SKILL.md must include non-empty frontmatter "
"name and description.",
)
return name


def _resolve_skill_test_dir(skill: str, agent_id: str) -> Path:
"""Resolve a skill argument as a path first, then workspace skill name."""
candidate = Path(skill).expanduser()
if candidate.exists():
return candidate.resolve()

working_dir = _get_agent_workspace(agent_id)
return get_workspace_skills_dir(working_dir) / skill


def _run_skill_test(skill_dir: Path) -> str:
"""Run local skill validation and security scanning."""
if not skill_dir.is_dir():
raise click.ClickException(f"Skill directory not found: {skill_dir}")

declared_name = _validate_skill_frontmatter(skill_dir)
try:
result = scan_skill_directory(
skill_dir,
skill_name=declared_name,
block=True,
)
except SkillScanError as exc:
raise click.ClickException(str(exc)) from exc

if result is not None and not result.is_safe:
raise click.ClickException(
"Security scan found "
f"{len(result.findings)} issue(s) in skill '{declared_name}'.",
)
return declared_name


def _apply_skill_changes(
skill_service: SkillService,
pool_service: SkillPoolService | None,
Expand Down Expand Up @@ -314,3 +399,18 @@ def info_cmd(
click.echo(
"Description: " f"{skill.description or 'No description.'}",
)


@skills_group.command("test")
@click.argument("skill", required=True)
@click.option(
"--agent-id",
default="default",
help="Agent ID (defaults to 'default')",
)
def test_cmd(skill: str, agent_id: str) -> None:
"""Validate a workspace skill or local skill directory."""
skill_dir = _resolve_skill_test_dir(skill, agent_id)
declared_name = _run_skill_test(skill_dir)
click.echo(f"Skill test passed: {declared_name}")
click.echo(f"Path: {skill_dir}")
102 changes: 102 additions & 0 deletions tests/unit/cli/test_cli_skills.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
# -*- coding: utf-8 -*-
"""Tests for qwenpaw skills CLI commands."""
from pathlib import Path
from unittest.mock import patch

from click.testing import CliRunner

from qwenpaw.cli.skills_cmd import skills_group


def _write_skill(skill_dir: Path, name: str = "demo") -> None:
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text(
f"""---
name: {name}
description: Demo skill
---

# Demo
""",
encoding="utf-8",
)


def test_skills_test_accepts_local_skill_directory(tmp_path):
"""Validate a skill passed as a local directory path."""
skill_dir = tmp_path / "demo"
_write_skill(skill_dir)

runner = CliRunner()
with patch("qwenpaw.cli.skills_cmd.scan_skill_directory") as mock_scan:
mock_scan.return_value = None
result = runner.invoke(skills_group, ["test", str(skill_dir)])

assert result.exit_code == 0
assert "Skill test passed: demo" in result.output
mock_scan.assert_called_once_with(
skill_dir.resolve(),
skill_name="demo",
block=True,
)


def test_skills_test_resolves_workspace_skill_name(tmp_path):
"""Validate a skill by name from the selected agent workspace."""
workspace_dir = tmp_path / "workspace"
skill_dir = workspace_dir / "skills" / "demo"
_write_skill(skill_dir)

runner = CliRunner()
with patch(
"qwenpaw.cli.skills_cmd._get_agent_workspace",
return_value=workspace_dir,
) as mock_workspace:
with patch("qwenpaw.cli.skills_cmd.scan_skill_directory") as mock_scan:
mock_scan.return_value = None
result = runner.invoke(
skills_group,
["test", "demo", "--agent-id", "agent-1"],
)

assert result.exit_code == 0
assert "Skill test passed: demo" in result.output
mock_workspace.assert_called_once_with("agent-1")
mock_scan.assert_called_once_with(
skill_dir,
skill_name="demo",
block=True,
)


def test_skills_test_fails_when_skill_md_is_missing(tmp_path):
"""Report missing SKILL.md as a command error."""
skill_dir = tmp_path / "demo"
skill_dir.mkdir()

runner = CliRunner()
result = runner.invoke(skills_group, ["test", str(skill_dir)])

assert result.exit_code != 0
assert "Missing SKILL.md" in result.output


def test_skills_test_fails_when_required_frontmatter_is_missing(tmp_path):
"""Require name and description in SKILL.md frontmatter."""
skill_dir = tmp_path / "demo"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text(
"""---
name: demo
---

# Demo
""",
encoding="utf-8",
)

runner = CliRunner()
result = runner.invoke(skills_group, ["test", str(skill_dir)])

assert result.exit_code != 0
assert "name and description" in result.output
Loading