forked from cleveragents/cleveragents-core
499 lines
17 KiB
Python
499 lines
17 KiB
Python
"""Step definitions for the Skill CLI feature.
|
|
|
|
Tests for features/skill_cli.feature — validates the ``agents skill``
|
|
command group: add, remove, list, show, and tools.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
|
|
import yaml
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands.skill import (
|
|
_get_skill_service,
|
|
_reset_skill_service,
|
|
)
|
|
from cleveragents.cli.commands.skill import (
|
|
app as skill_app,
|
|
)
|
|
from cleveragents.domain.models.core.skill import Skill, SkillInclude
|
|
|
|
# ────────────────────────────────────────────────────────────
|
|
# YAML fixtures
|
|
# ────────────────────────────────────────────────────────────
|
|
|
|
_SIMPLE_SKILL_YAML = """\
|
|
name: local/file-reader
|
|
description: "Basic file reading operations"
|
|
tools:
|
|
- name: builtin/read_file
|
|
- name: builtin/list_directory
|
|
- name: builtin/search_files
|
|
"""
|
|
|
|
_FULL_SKILL_YAML = """\
|
|
name: local/full-skill
|
|
description: "A fully populated skill"
|
|
tools:
|
|
- name: builtin/read_file
|
|
- name: builtin/write_file
|
|
writes: true
|
|
includes:
|
|
- name: local/file-reader
|
|
inline_tools:
|
|
- name: word_count
|
|
description: "Count words in text"
|
|
source: custom
|
|
code: |
|
|
def run(input_data):
|
|
return {"count": 0}
|
|
writes: false
|
|
mcp_servers:
|
|
- name: github
|
|
transport: stdio
|
|
command: npx
|
|
args: ["-y", "@modelcontextprotocol/server-github"]
|
|
env:
|
|
TOKEN: "test"
|
|
tool_filter:
|
|
include: [create_issue]
|
|
agent_skill_folders:
|
|
- path: ./skills/deploy
|
|
name: deploy
|
|
"""
|
|
|
|
_INVALID_SKILL_YAML = """\
|
|
not_a_name: missing required name field
|
|
extra_field: should fail validation
|
|
"""
|
|
|
|
_GIT_OPS_YAML = """\
|
|
name: local/git-ops
|
|
description: "Git operations"
|
|
tools:
|
|
- name: builtin/git_status
|
|
- name: builtin/git_diff
|
|
"""
|
|
|
|
_DEVOPS_YAML = """\
|
|
name: devops/deploy-tools
|
|
description: "Deployment tools"
|
|
tools:
|
|
- name: builtin/shell_execute
|
|
"""
|
|
|
|
_MCP_SKILL_YAML = """\
|
|
name: local/mcp-backed
|
|
description: "MCP backed skill"
|
|
mcp_servers:
|
|
- name: linear
|
|
transport: stdio
|
|
command: npx
|
|
args: ["-y", "@modelcontextprotocol/server-linear"]
|
|
env:
|
|
TOKEN: "test"
|
|
"""
|
|
|
|
|
|
# ────────────────────────────────────────────────────────────
|
|
# Helpers
|
|
# ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _write_temp_yaml(context: Context, content: str) -> str:
|
|
"""Write YAML content to a temporary file and register cleanup."""
|
|
fd, path = tempfile.mkstemp(suffix=".yaml")
|
|
with os.fdopen(fd, "w") as fh:
|
|
fh.write(content)
|
|
if not hasattr(context, "_skill_cleanup"):
|
|
context._skill_cleanup = []
|
|
context._skill_cleanup.append(path)
|
|
return path
|
|
|
|
|
|
def _register_skill_directly(
|
|
name: str,
|
|
description: str = "Test skill",
|
|
tool_refs: list[str] | None = None,
|
|
includes: list[str] | None = None,
|
|
mcp_servers: list[dict] | None = None,
|
|
) -> Skill:
|
|
"""Register a skill directly in the service for test setup."""
|
|
service = _get_skill_service()
|
|
skill = Skill(
|
|
name=name,
|
|
description=description,
|
|
tool_refs=tool_refs or [],
|
|
includes=[SkillInclude(name=n) for n in (includes or [])],
|
|
)
|
|
if mcp_servers:
|
|
from cleveragents.domain.models.core.skill import SkillMcpSource
|
|
|
|
skill = skill.model_copy(
|
|
update={
|
|
"mcp_servers": [
|
|
SkillMcpSource(server=m["server"], tools=m.get("tools"))
|
|
for m in mcp_servers
|
|
]
|
|
}
|
|
)
|
|
from datetime import datetime
|
|
|
|
now = datetime.now()
|
|
service._skills[name] = skill
|
|
service._created_at[name] = now
|
|
service._updated_at[name] = now
|
|
return skill
|
|
|
|
|
|
# ────────────────────────────────────────────────────────────
|
|
# Given steps
|
|
# ────────────────────────────────────────────────────────────
|
|
|
|
|
|
@given("a skill CLI test runner")
|
|
def step_skill_cli_runner(context: Context) -> None:
|
|
"""Set up the CLI runner for skill testing."""
|
|
context.skill_runner = CliRunner()
|
|
context.skill_result = None
|
|
context._skill_cleanup = []
|
|
|
|
|
|
@given("the skill service is reset")
|
|
def step_skill_service_reset(context: Context) -> None:
|
|
"""Reset the skill service to empty state."""
|
|
_reset_skill_service()
|
|
|
|
|
|
@given("a valid skill config YAML file at a temp path")
|
|
def step_valid_skill_yaml(context: Context) -> None:
|
|
"""Write a simple valid skill YAML to a temp file."""
|
|
context.skill_yaml_path = _write_temp_yaml(context, _SIMPLE_SKILL_YAML)
|
|
|
|
|
|
@given("a full skill config YAML file at a temp path")
|
|
def step_full_skill_yaml(context: Context) -> None:
|
|
"""Write a fully-populated skill YAML to a temp file."""
|
|
context.skill_yaml_path = _write_temp_yaml(context, _FULL_SKILL_YAML)
|
|
|
|
|
|
@given("an invalid skill YAML file at a temp path")
|
|
def step_invalid_skill_yaml(context: Context) -> None:
|
|
"""Write invalid YAML to a temp file."""
|
|
context.skill_yaml_path = _write_temp_yaml(context, _INVALID_SKILL_YAML)
|
|
|
|
|
|
@given('the skill "{name}" is already registered')
|
|
def step_skill_already_registered(context: Context, name: str) -> None:
|
|
"""Register a skill directly so it already exists."""
|
|
_register_skill_directly(
|
|
name,
|
|
description="Basic file reading operations",
|
|
tool_refs=[
|
|
"builtin/read_file",
|
|
"builtin/list_directory",
|
|
"builtin/search_files",
|
|
],
|
|
)
|
|
|
|
|
|
@given('the skill "{name}" is registered with tools')
|
|
def step_skill_registered_with_tools(context: Context, name: str) -> None:
|
|
"""Register a skill with some tools."""
|
|
if "git-ops" in name:
|
|
_register_skill_directly(
|
|
name,
|
|
description="Git operations",
|
|
tool_refs=["builtin/git_status", "builtin/git_diff"],
|
|
)
|
|
elif "deploy" in name:
|
|
_register_skill_directly(
|
|
name,
|
|
description="Deployment tools",
|
|
tool_refs=["builtin/shell_execute"],
|
|
)
|
|
else:
|
|
_register_skill_directly(
|
|
name,
|
|
description="Basic file reading operations",
|
|
tool_refs=[
|
|
"builtin/read_file",
|
|
"builtin/list_directory",
|
|
"builtin/search_files",
|
|
],
|
|
)
|
|
|
|
|
|
@given('a composed skill "{name}" including "{included}" is registered')
|
|
def step_composed_skill_registered(context: Context, name: str, included: str) -> None:
|
|
"""Register a composed skill that includes another."""
|
|
# First ensure the included skill is registered
|
|
service = _get_skill_service()
|
|
if included not in service._skills:
|
|
_register_skill_directly(
|
|
included,
|
|
description="Included skill",
|
|
tool_refs=["builtin/read_file"],
|
|
)
|
|
_register_skill_directly(
|
|
name,
|
|
description="Composed skill",
|
|
tool_refs=["builtin/shell_execute"],
|
|
includes=[included],
|
|
)
|
|
|
|
|
|
@given('a skill "{name}" with MCP servers is registered')
|
|
def step_skill_with_mcp_registered(context: Context, name: str) -> None:
|
|
"""Register a skill with MCP servers."""
|
|
_register_skill_directly(
|
|
name,
|
|
description="MCP backed skill",
|
|
mcp_servers=[{"server": "linear", "tools": ["create_issue"]}],
|
|
)
|
|
|
|
|
|
@given('a skill "{name}" including "{included}" is registered')
|
|
def step_skill_including_registered(context: Context, name: str, included: str) -> None:
|
|
"""Register a skill that includes another (may create cycle)."""
|
|
service = _get_skill_service()
|
|
if included not in service._skills:
|
|
_register_skill_directly(included, description="Placeholder")
|
|
_register_skill_directly(
|
|
name,
|
|
description="Skill with include",
|
|
includes=[included],
|
|
)
|
|
|
|
|
|
# ────────────────────────────────────────────────────────────
|
|
# When steps
|
|
# ────────────────────────────────────────────────────────────
|
|
|
|
|
|
@when("I run skill CLI add with --config pointing to the YAML file")
|
|
def step_run_skill_add(context: Context) -> None:
|
|
"""Run ``skill add --config <file>``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["add", "--config", context.skill_yaml_path]
|
|
)
|
|
|
|
|
|
@when("I run skill CLI add with --config and --update pointing to the YAML file")
|
|
def step_run_skill_add_update(context: Context) -> None:
|
|
"""Run ``skill add --config <file> --update``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["add", "--config", context.skill_yaml_path, "--update"]
|
|
)
|
|
|
|
|
|
@when("I run skill CLI add with --config pointing to a missing file")
|
|
def step_run_skill_add_missing(context: Context) -> None:
|
|
"""Run ``skill add --config /nonexistent/path.yaml``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["add", "--config", "/nonexistent/path.yaml"]
|
|
)
|
|
|
|
|
|
@when("I run skill CLI add with --config and --format json")
|
|
def step_run_skill_add_json(context: Context) -> None:
|
|
"""Run ``skill add --config <file> --format json``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["add", "--config", context.skill_yaml_path, "--format", "json"]
|
|
)
|
|
|
|
|
|
@when("I run skill CLI add with --config and --format yaml")
|
|
def step_run_skill_add_yaml(context: Context) -> None:
|
|
"""Run ``skill add --config <file> --format yaml``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["add", "--config", context.skill_yaml_path, "--format", "yaml"]
|
|
)
|
|
|
|
|
|
@when('I run skill CLI show "{name}"')
|
|
def step_run_skill_show(context: Context, name: str) -> None:
|
|
"""Run ``skill show <name>``."""
|
|
context.skill_result = context.skill_runner.invoke(skill_app, ["show", name])
|
|
|
|
|
|
@when('I run skill CLI show "{name}" with --format json')
|
|
def step_run_skill_show_json(context: Context, name: str) -> None:
|
|
"""Run ``skill show <name> --format json``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["show", name, "--format", "json"]
|
|
)
|
|
|
|
|
|
@when('I run skill CLI tools "{name}"')
|
|
def step_run_skill_tools(context: Context, name: str) -> None:
|
|
"""Run ``skill tools <name>``."""
|
|
context.skill_result = context.skill_runner.invoke(skill_app, ["tools", name])
|
|
|
|
|
|
@when('I run skill CLI tools "{name}" with --format json')
|
|
def step_run_skill_tools_json(context: Context, name: str) -> None:
|
|
"""Run ``skill tools <name> --format json``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["tools", name, "--format", "json"]
|
|
)
|
|
|
|
|
|
@when("I run skill CLI list")
|
|
def step_run_skill_list(context: Context) -> None:
|
|
"""Run ``skill list``."""
|
|
context.skill_result = context.skill_runner.invoke(skill_app, ["list"])
|
|
|
|
|
|
@when('I run skill CLI list with --namespace "{ns}"')
|
|
def step_run_skill_list_namespace(context: Context, ns: str) -> None:
|
|
"""Run ``skill list --namespace <ns>``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["list", "--namespace", ns]
|
|
)
|
|
|
|
|
|
@when('I run skill CLI list with --source "{source}"')
|
|
def step_run_skill_list_source(context: Context, source: str) -> None:
|
|
"""Run ``skill list --source <source>``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["list", "--source", source]
|
|
)
|
|
|
|
|
|
@when("I run skill CLI list with --format json")
|
|
def step_run_skill_list_json(context: Context) -> None:
|
|
"""Run ``skill list --format json``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["list", "--format", "json"]
|
|
)
|
|
|
|
|
|
@when('I run skill CLI remove "{name}" with --yes')
|
|
def step_run_skill_remove_yes(context: Context, name: str) -> None:
|
|
"""Run ``skill remove <name> --yes``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["remove", name, "--yes"]
|
|
)
|
|
|
|
|
|
@when('I run skill CLI remove "{name}" with --yes and --format json')
|
|
def step_run_skill_remove_json(context: Context, name: str) -> None:
|
|
"""Run ``skill remove <name> --yes --format json``."""
|
|
context.skill_result = context.skill_runner.invoke(
|
|
skill_app, ["remove", name, "--yes", "--format", "json"]
|
|
)
|
|
|
|
|
|
# ────────────────────────────────────────────────────────────
|
|
# Then steps
|
|
# ────────────────────────────────────────────────────────────
|
|
|
|
|
|
@then("the skill CLI add should succeed")
|
|
def step_skill_add_succeed(context: Context) -> None:
|
|
"""Assert ``skill add`` exited with code 0."""
|
|
assert context.skill_result is not None
|
|
assert context.skill_result.exit_code == 0, (
|
|
f"Expected exit_code 0, got {context.skill_result.exit_code}.\n"
|
|
f"Output: {context.skill_result.output}"
|
|
)
|
|
|
|
|
|
@then("the skill CLI show should succeed")
|
|
def step_skill_show_succeed(context: Context) -> None:
|
|
"""Assert ``skill show`` exited with code 0."""
|
|
assert context.skill_result is not None
|
|
assert context.skill_result.exit_code == 0, (
|
|
f"Expected exit_code 0, got {context.skill_result.exit_code}.\n"
|
|
f"Output: {context.skill_result.output}"
|
|
)
|
|
|
|
|
|
@then("the skill CLI tools should succeed")
|
|
def step_skill_tools_succeed(context: Context) -> None:
|
|
"""Assert ``skill tools`` exited with code 0."""
|
|
assert context.skill_result is not None
|
|
assert context.skill_result.exit_code == 0, (
|
|
f"Expected exit_code 0, got {context.skill_result.exit_code}.\n"
|
|
f"Output: {context.skill_result.output}"
|
|
)
|
|
|
|
|
|
@then("the skill CLI list should succeed")
|
|
def step_skill_list_succeed(context: Context) -> None:
|
|
"""Assert ``skill list`` exited with code 0."""
|
|
assert context.skill_result is not None
|
|
assert context.skill_result.exit_code == 0, (
|
|
f"Expected exit_code 0, got {context.skill_result.exit_code}.\n"
|
|
f"Output: {context.skill_result.output}"
|
|
)
|
|
|
|
|
|
@then("the skill CLI remove should succeed")
|
|
def step_skill_remove_succeed(context: Context) -> None:
|
|
"""Assert ``skill remove`` exited with code 0."""
|
|
assert context.skill_result is not None
|
|
assert context.skill_result.exit_code == 0, (
|
|
f"Expected exit_code 0, got {context.skill_result.exit_code}.\n"
|
|
f"Output: {context.skill_result.output}"
|
|
)
|
|
|
|
|
|
@then("the skill CLI command should abort")
|
|
def step_skill_command_abort(context: Context) -> None:
|
|
"""Assert the skill command exited with non-zero."""
|
|
assert context.skill_result is not None
|
|
assert context.skill_result.exit_code != 0, (
|
|
f"Expected non-zero exit, got 0.\nOutput: {context.skill_result.output}"
|
|
)
|
|
|
|
|
|
@then('the skill CLI output should contain "{text}"')
|
|
def step_skill_output_contains(context: Context, text: str) -> None:
|
|
"""Assert the output contains the expected text."""
|
|
assert context.skill_result is not None
|
|
output = context.skill_result.output
|
|
assert text in output, f"Expected '{text}' in output.\nOutput:\n{output}"
|
|
|
|
|
|
@then('the skill CLI output should not contain "{text}"')
|
|
def step_skill_output_not_contains(context: Context, text: str) -> None:
|
|
"""Assert the output does NOT contain the text."""
|
|
assert context.skill_result is not None
|
|
output = context.skill_result.output
|
|
assert text not in output, f"Did not expect '{text}' in output.\nOutput:\n{output}"
|
|
|
|
|
|
@then("the skill CLI output should be valid JSON")
|
|
def step_skill_output_valid_json(context: Context) -> None:
|
|
"""Assert the output is valid JSON."""
|
|
assert context.skill_result is not None
|
|
output = context.skill_result.output.strip()
|
|
try:
|
|
json.loads(output)
|
|
except json.JSONDecodeError as exc:
|
|
raise AssertionError(
|
|
f"Output is not valid JSON: {exc}\nOutput:\n{output}"
|
|
) from exc
|
|
|
|
|
|
@then("the skill CLI output should be valid YAML")
|
|
def step_skill_output_valid_yaml(context: Context) -> None:
|
|
"""Assert the output is valid YAML."""
|
|
assert context.skill_result is not None
|
|
output = context.skill_result.output.strip()
|
|
try:
|
|
result = yaml.safe_load(output)
|
|
assert result is not None
|
|
except yaml.YAMLError as exc:
|
|
raise AssertionError(
|
|
f"Output is not valid YAML: {exc}\nOutput:\n{output}"
|
|
) from exc
|