cleveragents-core/robot/helper_skill_cli.py

"""Helper script for skill_cli.robot smoke tests.

Each subcommand is a self-contained check that prints a sentinel on success.
"""

from __future__ import annotations

import os
import sys
import tempfile
from datetime import datetime
from pathlib import Path

# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from typer.testing import CliRunner  # noqa: E402

from cleveragents.application.services.skill_service import SkillService  # noqa: E402
from cleveragents.cli.commands.skill import (  # noqa: E402
    _reset_skill_service,
)
from cleveragents.cli.commands.skill import (  # noqa: E402
    app as skill_app,
)
from cleveragents.domain.models.core.skill import Skill  # noqa: E402

runner = CliRunner()

_VALID_YAML = """\
name: local/smoke-skill
description: "Smoke test skill"
tools:
  - name: builtin/read_file
  - name: builtin/list_directory
"""

_COMPOSED_YAML = """\
name: local/composed-smoke
description: "Composed smoke test skill"
tools:
  - name: builtin/shell_execute
includes:
  - name: local/smoke-skill
"""


def _write_yaml(content: str) -> str:
    fd, path = tempfile.mkstemp(suffix=".yaml")
    with os.fdopen(fd, "w") as fh:
        fh.write(content)
    return path


def _fresh_service() -> SkillService:
    """Create and install a fresh SkillService."""
    svc = SkillService()
    _reset_skill_service(svc)
    return svc


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def add_config() -> None:
    """Verify skill add --config works."""
    _fresh_service()
    path = _write_yaml(_VALID_YAML)
    try:
        result = runner.invoke(skill_app, ["add", "--config", path])
        if result.exit_code == 0 and "Skill Registered" in result.output:
            print("skill-cli-add-config-ok")
        else:
            print(f"FAIL: add returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)
    finally:
        Path(path).unlink(missing_ok=True)


def add_duplicate_fails() -> None:
    """Verify that adding a duplicate skill fails without --update."""
    svc = _fresh_service()
    # Pre-register the skill
    skill = Skill(
        name="local/smoke-skill",
        description="Smoke test skill",
        tool_refs=["builtin/read_file"],
    )
    svc._skills["local/smoke-skill"] = skill
    svc._created_at["local/smoke-skill"] = datetime.now()
    svc._updated_at["local/smoke-skill"] = datetime.now()

    path = _write_yaml(_VALID_YAML)
    try:
        result = runner.invoke(skill_app, ["add", "--config", path])
        if result.exit_code != 0 and "already registered" in result.output:
            print("skill-cli-add-duplicate-fails-ok")
        else:
            print(f"FAIL: expected error, got {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)
    finally:
        Path(path).unlink(missing_ok=True)


def add_update() -> None:
    """Verify skill add --update works."""
    svc = _fresh_service()
    # Pre-register
    skill = Skill(
        name="local/smoke-skill",
        description="Old description",
        tool_refs=["builtin/read_file"],
    )
    svc._skills["local/smoke-skill"] = skill
    svc._created_at["local/smoke-skill"] = datetime.now()
    svc._updated_at["local/smoke-skill"] = datetime.now()

    path = _write_yaml(_VALID_YAML)
    try:
        result = runner.invoke(skill_app, ["add", "--config", path, "--update"])
        if result.exit_code == 0 and "Skill Updated" in result.output:
            print("skill-cli-add-update-ok")
        else:
            print(f"FAIL: update returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)
    finally:
        Path(path).unlink(missing_ok=True)


def show_skill() -> None:
    """Verify skill show works."""
    svc = _fresh_service()
    skill = Skill(
        name="local/smoke-skill",
        description="Smoke test skill",
        tool_refs=["builtin/read_file", "builtin/list_directory"],
    )
    svc._skills["local/smoke-skill"] = skill
    svc._created_at["local/smoke-skill"] = datetime.now()
    svc._updated_at["local/smoke-skill"] = datetime.now()

    result = runner.invoke(skill_app, ["show", "local/smoke-skill"])
    if result.exit_code == 0 and "Skill Details" in result.output:
        print("skill-cli-show-ok")
    else:
        print(f"FAIL: show returned {result.exit_code}", file=sys.stderr)
        print(result.output, file=sys.stderr)
        sys.exit(1)


def tools_skill() -> None:
    """Verify skill tools works."""
    svc = _fresh_service()
    skill = Skill(
        name="local/smoke-skill",
        description="Smoke test skill",
        tool_refs=["builtin/read_file", "builtin/list_directory"],
    )
    svc._skills["local/smoke-skill"] = skill
    svc._created_at["local/smoke-skill"] = datetime.now()
    svc._updated_at["local/smoke-skill"] = datetime.now()

    result = runner.invoke(skill_app, ["tools", "local/smoke-skill"])
    if result.exit_code == 0 and "Tools for" in result.output:
        print("skill-cli-tools-ok")
    else:
        print(f"FAIL: tools returned {result.exit_code}", file=sys.stderr)
        print(result.output, file=sys.stderr)
        sys.exit(1)


def list_skills() -> None:
    """Verify skill list works."""
    svc = _fresh_service()
    for name in ["local/skill-a", "local/skill-b"]:
        skill = Skill(
            name=name,
            description=f"Test skill {name}",
            tool_refs=["builtin/read_file"],
        )
        svc._skills[name] = skill
        svc._created_at[name] = datetime.now()
        svc._updated_at[name] = datetime.now()

    result = runner.invoke(skill_app, ["list"])
    if result.exit_code == 0 and "Summary" in result.output:
        print("skill-cli-list-ok")
    else:
        print(f"FAIL: list returned {result.exit_code}", file=sys.stderr)
        print(result.output, file=sys.stderr)
        sys.exit(1)


def remove_skill() -> None:
    """Verify skill remove --yes works."""
    svc = _fresh_service()
    skill = Skill(
        name="local/smoke-skill",
        description="Smoke test skill",
        tool_refs=["builtin/read_file"],
    )
    svc._skills["local/smoke-skill"] = skill
    svc._created_at["local/smoke-skill"] = datetime.now()
    svc._updated_at["local/smoke-skill"] = datetime.now()

    result = runner.invoke(skill_app, ["remove", "local/smoke-skill", "--yes"])
    if result.exit_code == 0 and "Skill Removed" in result.output:
        print("skill-cli-remove-ok")
    else:
        print(f"FAIL: remove returned {result.exit_code}", file=sys.stderr)
        print(result.output, file=sys.stderr)
        sys.exit(1)


def refresh_skill() -> None:
    """Verify skill refresh <name> works."""
    svc = _fresh_service()
    skill = Skill(
        name="local/smoke-skill",
        description="Smoke test skill",
        tool_refs=["builtin/read_file", "builtin/list_directory"],
    )
    svc._skills["local/smoke-skill"] = skill
    svc._created_at["local/smoke-skill"] = datetime.now()
    svc._updated_at["local/smoke-skill"] = datetime.now()

    result = runner.invoke(skill_app, ["refresh", "local/smoke-skill"])
    if result.exit_code == 0 and "Skill Refreshed" in result.output:
        print("skill-cli-refresh-ok")
    else:
        print(f"FAIL: refresh returned {result.exit_code}", file=sys.stderr)
        print(result.output, file=sys.stderr)
        sys.exit(1)


def refresh_all() -> None:
    """Verify skill refresh --all works."""
    svc = _fresh_service()
    skill1 = Skill(
        name="local/smoke-skill",
        description="Smoke test skill",
        tool_refs=["builtin/read_file"],
    )
    skill2 = Skill(
        name="local/another-skill",
        description="Another skill",
        tool_refs=["builtin/shell_execute"],
    )
    svc._skills["local/smoke-skill"] = skill1
    svc._created_at["local/smoke-skill"] = datetime.now()
    svc._updated_at["local/smoke-skill"] = datetime.now()
    svc._skills["local/another-skill"] = skill2
    svc._created_at["local/another-skill"] = datetime.now()
    svc._updated_at["local/another-skill"] = datetime.now()

    result = runner.invoke(skill_app, ["refresh", "--all"])
    if result.exit_code == 0 and "Skills Refreshed" in result.output:
        print("skill-cli-refresh-all-ok")
    else:
        print(f"FAIL: refresh --all returned {result.exit_code}", file=sys.stderr)
        print(result.output, file=sys.stderr)
        sys.exit(1)


# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------

_COMMANDS = {
    "add-config": add_config,
    "add-duplicate-fails": add_duplicate_fails,
    "add-update": add_update,
    "show-skill": show_skill,
    "tools-skill": tools_skill,
    "list-skills": list_skills,
    "remove-skill": remove_skill,
    "refresh-skill": refresh_skill,
    "refresh-all": refresh_all,
}


def main() -> None:
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
        sys.exit(2)
    _COMMANDS[sys.argv[1]]()


if __name__ == "__main__":
    main()