cleveragents-core/features/steps/skill_cli_steps.py

"""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,
    SkillAgentSource,
    SkillInclude,
    SkillInlineTool,
    SkillMcpSource,
)
from cleveragents.domain.models.core.tool import ToolCapability, ToolSource

# ────────────────────────────────────────────────────────────
# 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"
"""

_NO_DESCRIPTION_YAML = """\
name: local/no-desc
tools:
  - name: builtin/echo
"""


# ────────────────────────────────────────────────────────────
# 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


# ────────────────────────────────────────────────────────────
# Coverage - Given steps for richer skill configurations
# ────────────────────────────────────────────────────────────


@given('a skill "{name}" with MCP servers and tools is registered')
def step_skill_mcp_with_tools(context: Context, name: str) -> None:
    """Register a skill that has MCP servers with explicit tool lists."""
    service = _get_skill_service()
    skill = Skill(
        name=name,
        description="MCP skill with tools",
        mcp_servers=[
            SkillMcpSource(server="github-mcp", tools=["create_issue", "list_prs"]),
        ],
    )
    from datetime import datetime

    now = datetime.now()
    service._skills[name] = skill
    service._created_at[name] = now
    service._updated_at[name] = now


@given('a skill "{name}" with agent_skills and inline tools is registered')
def step_skill_agent_and_inline(context: Context, name: str) -> None:
    """Register a skill with agent_skills and anonymous inline tools."""
    service = _get_skill_service()
    skill = Skill(
        name=name,
        description="Mixed sources skill",
        tool_refs=["builtin/read_file"],
        agent_skills=[SkillAgentSource(path="./skills/deploy")],
        anonymous_tools=[
            SkillInlineTool(
                description="Inline formatter",
                source=ToolSource.CUSTOM,
                code="return formatted",
                timeout=300,
                capability=ToolCapability(
                    read_only=False,
                    writes=True,
                    checkpointable=True,
                ),
            ),
        ],
    )
    from datetime import datetime

    now = datetime.now()
    service._skills[name] = skill
    service._created_at[name] = now
    service._updated_at[name] = now


@given('a skill "{name}" with MCP servers and inline tools is registered')
def step_skill_mcp_and_inline(context: Context, name: str) -> None:
    """Register a skill with MCP servers and inline tools for tools cmd."""
    service = _get_skill_service()
    skill = Skill(
        name=name,
        description="MCP and inline skill",
        mcp_servers=[
            SkillMcpSource(server="test-server", tools=["mcp-tool-a"]),
        ],
        anonymous_tools=[
            SkillInlineTool(
                description="Inline tool",
                source=ToolSource.CUSTOM,
                code="return 1",
                timeout=300,
                capability=ToolCapability(
                    read_only=True,
                    writes=False,
                    checkpointable=False,
                ),
            ),
        ],
    )
    from datetime import datetime

    now = datetime.now()
    service._skills[name] = skill
    service._created_at[name] = now
    service._updated_at[name] = now


@given('a skill "{name}" with only tool_refs is registered')
def step_skill_only_tool_refs(context: Context, name: str) -> None:
    """Register a skill with only builtin tool_refs."""
    _register_skill_directly(
        name, description="Builtin only", tool_refs=["builtin/echo"]
    )


@given('a skill "{name}" with agent_skills is registered')
def step_skill_with_agent_skills(context: Context, name: str) -> None:
    """Register a skill with agent_skill sources."""
    service = _get_skill_service()
    skill = Skill(
        name=name,
        description="Agent skill source",
        agent_skills=[SkillAgentSource(path="./skills/agent-folder")],
    )
    from datetime import datetime

    now = datetime.now()
    service._skills[name] = skill
    service._created_at[name] = now
    service._updated_at[name] = now


@given("a skill is registered without a config path")
def step_skill_no_config_path(context: Context) -> None:
    """Register a skill via add_skill with config_path=None."""
    from cleveragents.skills.schema import SkillConfigSchema

    service = _get_skill_service()
    schema = SkillConfigSchema(name="local/no-path", description="No path skill")
    service.add_skill(schema, config_path=None)
    context.no_path_skill_name = "local/no-path"


@given("a skill config YAML with no description at a temp path")
def step_no_desc_yaml(context: Context) -> None:
    """Write a skill YAML with no description to a temp file."""
    context.skill_yaml_path = _write_temp_yaml(context, _NO_DESCRIPTION_YAML)


# ────────────────────────────────────────────────────────────
# Coverage - When steps
# ────────────────────────────────────────────────────────────


@when('I run skill CLI remove "{name}" without --yes and deny')
def step_run_skill_remove_deny(context: Context, name: str) -> None:
    """Run ``skill remove <name>`` without --yes and deny the prompt."""
    context.skill_result = context.skill_runner.invoke(
        skill_app, ["remove", name], input="n\n"
    )


@when('I list skills with source "{source}"')
def step_list_skills_source(context: Context, source: str) -> None:
    """List skills filtered by source type via the service."""
    service = _get_skill_service()
    context.skill_list_result = service.list_skills(source=source)


# ────────────────────────────────────────────────────────────
# Coverage - Then steps
# ────────────────────────────────────────────────────────────


@then("the skill service should report skill count {count:d}")
def step_skill_count(context: Context, count: int) -> None:
    """Assert the service skill_count matches."""
    service = _get_skill_service()
    actual = service.skill_count()
    assert actual == count, f"Expected {count}, got {actual}"


@then("the config path should be empty for the skill")
def step_config_path_empty(context: Context) -> None:
    """Assert no config path is stored."""
    service = _get_skill_service()
    path = service.get_config_path(context.no_path_skill_name)
    assert path is None, f"Expected None config path, got {path}"


@then('the skill list should contain "{name}"')
def step_skill_list_contains(context: Context, name: str) -> None:
    """Assert the filtered skill list contains the given name."""
    names = [s.name for s in context.skill_list_result]
    assert name in names, f"Expected '{name}' in {names}"


@then("removing a skill with empty name should raise ValueError")
def step_remove_empty_name(context: Context) -> None:
    """Assert remove_skill('') raises ValueError."""
    service = _get_skill_service()
    try:
        service.remove_skill("")
        raise AssertionError("Expected ValueError")
    except ValueError:
        pass


@then('removing skill "{name}" should raise KeyError')
def step_remove_nonexistent(context: Context, name: str) -> None:
    """Assert remove_skill for nonexistent skill raises KeyError."""
    service = _get_skill_service()
    try:
        service.remove_skill(name)
        raise AssertionError("Expected KeyError")
    except KeyError:
        pass


@then("getting a skill with empty name should raise ValueError")
def step_get_empty_name(context: Context) -> None:
    """Assert get_skill('') raises ValueError."""
    service = _get_skill_service()
    try:
        service.get_skill("")
        raise AssertionError("Expected ValueError")
    except ValueError:
        pass


# ────────────────────────────────────────────────────────────
# Refresh command steps
# ────────────────────────────────────────────────────────────


@when('I run skill CLI refresh "{name}"')
def step_run_skill_refresh(context: Context, name: str) -> None:
    """Run ``skill refresh <name>``."""
    context.skill_result = context.skill_runner.invoke(skill_app, ["refresh", name])


@when('I run skill CLI refresh "{name}" with --format json')
def step_run_skill_refresh_json(context: Context, name: str) -> None:
    """Run ``skill refresh <name> --format json``."""
    context.skill_result = context.skill_runner.invoke(
        skill_app, ["refresh", name, "--format", "json"]
    )


@when("I run skill CLI refresh with --all")
def step_run_skill_refresh_all(context: Context) -> None:
    """Run ``skill refresh --all``."""
    context.skill_result = context.skill_runner.invoke(skill_app, ["refresh", "--all"])


@when('I run skill CLI refresh "{name}" with --all')
def step_run_skill_refresh_both(context: Context, name: str) -> None:
    """Run ``skill refresh <name> --all`` (should fail)."""
    context.skill_result = context.skill_runner.invoke(
        skill_app, ["refresh", name, "--all"]
    )


@when("I run skill CLI refresh without arguments")
def step_run_skill_refresh_no_args(context: Context) -> None:
    """Run ``skill refresh`` (should fail)."""
    context.skill_result = context.skill_runner.invoke(skill_app, ["refresh"])


@then("the skill CLI refresh should succeed")
def step_skill_refresh_succeed(context: Context) -> None:
    """Assert ``skill refresh`` 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}"
    )


# ────────────────────────────────────────────────────────────
# Enhanced JSON output validation
# ────────────────────────────────────────────────────────────


@then('the skill CLI JSON output should have field "{field}"')
def step_json_has_field(context: Context, field: str) -> None:
    """Assert JSON output contains a top-level field."""
    assert context.skill_result is not None
    output = context.skill_result.output.strip()
    try:
        data = json.loads(output)
        assert field in data, (
            f"Expected field '{field}' in JSON output.\n"
            f"Keys: {list(data.keys())}\nOutput:\n{output}"
        )
    except json.JSONDecodeError as exc:
        raise AssertionError(
            f"Output is not valid JSON: {exc}\nOutput:\n{output}"
        ) from exc


@then('the skill CLI JSON output should have field "{field}" in first skill')
def step_json_has_field_in_first_skill(context: Context, field: str) -> None:
    """Assert JSON output's first list item contains a field."""
    assert context.skill_result is not None
    output = context.skill_result.output.strip()
    try:
        data = json.loads(output)
        assert isinstance(data, list) and len(data) > 0, (
            f"Expected JSON output to be a non-empty list.\nOutput:\n{output}"
        )
        first_skill = data[0]
        assert field in first_skill, (
            f"Expected field '{field}' in first skill.\n"
            f"Keys: {list(first_skill.keys())}\nOutput:\n{output}"
        )
    except json.JSONDecodeError as exc:
        raise AssertionError(
            f"Output is not valid JSON: {exc}\nOutput:\n{output}"
        ) from exc