cleveragents-core/robot/helper_skill_registry.py

"""Helper script for Robot Framework skill registry smoke tests.

Usage:
    python helper_skill_registry.py <command>

Commands:
    register-and-get     Register a skill and retrieve it
    list-filter          Register skills and list with namespace filter
    update-skill         Register and update a skill
    duplicate-reject     Verify duplicate skill names are rejected
    remove-skill         Register and remove a skill
"""

from __future__ import annotations

import sys
from pathlib import Path

# Ensure src is on the path
src_dir = str(Path(__file__).resolve().parents[1] / "src")
if src_dir not in sys.path:
    sys.path.insert(0, src_dir)

from sqlalchemy import create_engine, event, text  # noqa: E402
from sqlalchemy.orm import Session, sessionmaker  # noqa: E402

from cleveragents.application.services.skill_registry_service import (  # noqa: E402
    SkillRegistryService,
)
from cleveragents.domain.models.core.skill import Skill  # noqa: E402
from cleveragents.infrastructure.database.models import Base  # noqa: E402
from cleveragents.infrastructure.database.repositories import (  # noqa: E402
    DuplicateSkillError,
    SkillRepository,
)


def _enable_fk_pragma(dbapi_conn, _connection_record):  # type: ignore[no-untyped-def]
    """Enable SQLite foreign key enforcement per connection."""
    cursor = dbapi_conn.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()


def _setup() -> tuple[SkillRegistryService, sessionmaker[Session]]:
    """Create an in-memory DB and return the service + session factory."""
    engine = create_engine("sqlite:///:memory:", echo=False)
    event.listen(engine, "connect", _enable_fk_pragma)
    Base.metadata.create_all(engine)
    factory: sessionmaker[Session] = sessionmaker(bind=engine)
    repo = SkillRepository(session_factory=factory)
    svc = SkillRegistryService(skill_repo=repo)
    return svc, factory


def _commit(factory: sessionmaker[Session]) -> None:
    """Commit the pending transaction on the shared connection."""
    session = factory()
    try:
        session.execute(text("SELECT 1"))
        session.commit()
    finally:
        session.close()


def _make_skill(name: str, description: str = "Test skill") -> Skill:
    return Skill(name=name, description=description)


def cmd_register_and_get() -> None:
    svc, factory = _setup()
    skill = _make_skill("local/test-tool", "A test skill")
    svc.add_skill(skill)
    _commit(factory)

    result = svc.get_skill("local/test-tool")
    assert result is not None, "Skill not found"
    assert result.name == "local/test-tool"
    print("skill-registry-ok")


def cmd_list_filter() -> None:
    svc, factory = _setup()
    svc.add_skill(_make_skill("local/tool-a", "Tool A"))
    svc.add_skill(_make_skill("local/tool-b", "Tool B"))
    svc.add_skill(_make_skill("devops/deploy", "Deploy"))
    _commit(factory)

    all_skills = svc.list_skills()
    assert len(all_skills) == 3

    local_skills = svc.list_skills(namespace="local")
    assert len(local_skills) == 2

    print("skill-list-ok")


def cmd_update_skill() -> None:
    svc, factory = _setup()
    svc.add_skill(_make_skill("local/updatable", "Original"))
    _commit(factory)

    updated = _make_skill("local/updatable", "Updated desc")
    svc.update_skill(updated)
    _commit(factory)

    result = svc.get_skill("local/updatable")
    assert result is not None
    assert result.description == "Updated desc"
    print("skill-update-ok")


def cmd_duplicate_reject() -> None:
    svc, factory = _setup()
    svc.add_skill(_make_skill("local/dup-test", "First"))
    _commit(factory)

    try:
        svc.add_skill(_make_skill("local/dup-test", "Second"))
        _commit(factory)
        print("FAIL: should have raised DuplicateSkillError")
        sys.exit(1)
    except DuplicateSkillError:
        print("skill-duplicate-reject-ok")
    except Exception as exc:
        # tenacity may wrap the error; unwrap if needed
        cause = exc.__cause__ or exc
        if isinstance(cause, DuplicateSkillError):
            print("skill-duplicate-reject-ok")
        else:
            print(f"FAIL: unexpected {type(exc).__name__}: {exc}")
            sys.exit(1)


def cmd_remove_skill() -> None:
    svc, factory = _setup()
    svc.add_skill(_make_skill("local/removable", "Removable"))
    _commit(factory)

    result = svc.remove_skill("local/removable")
    _commit(factory)
    assert result is True

    check = svc.get_skill("local/removable")
    assert check is None
    print("skill-remove-ok")


COMMANDS = {
    "register-and-get": cmd_register_and_get,
    "list-filter": cmd_list_filter,
    "update-skill": cmd_update_skill,
    "duplicate-reject": cmd_duplicate_reject,
    "remove-skill": cmd_remove_skill,
}


if __name__ == "__main__":
    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(1)
    COMMANDS[sys.argv[1]]()