cleveragents-core/robot/helper_skill_add_persist.py

"""Helper script for Robot Framework skill add persistence smoke tests.

Usage:
    python helper_skill_add_persist.py <command>

Commands:
    persist-round-trip    Add a skill, recreate service, verify visible
    persist-multiple      Add two skills, recreate service, verify both
    persist-remove        Add skill, remove, recreate service, verify gone
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

# 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  # noqa: E402
from sqlalchemy.orm import Session, sessionmaker  # noqa: E402

from cleveragents.application.services.skill_service import (  # noqa: E402
    SkillService,
)
from cleveragents.infrastructure.database.models import Base  # noqa: E402
from cleveragents.infrastructure.database.repositories import (  # noqa: E402
    SkillRepository,
)
from cleveragents.skills.schema import SkillConfigSchema  # noqa: E402


def _enable_fk(dbapi_conn: Any, _rec: Any) -> None:
    cursor = dbapi_conn.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()


def _setup() -> tuple[sessionmaker[Session], Any]:
    engine = create_engine("sqlite:///:memory:", echo=False)
    event.listen(engine, "connect", _enable_fk)
    Base.metadata.create_all(engine)
    factory: sessionmaker[Session] = sessionmaker(bind=engine)
    return factory, engine


def _make_service(factory: Any) -> SkillService:
    shared = factory()

    def _sf() -> Session:
        return shared

    repo = SkillRepository(session_factory=_sf)
    return SkillService(skill_repo=repo, session_factory=_sf)


def _make_config(name: str) -> SkillConfigSchema:
    import os
    import tempfile

    yaml_text = (
        f"name: {name}\n"
        f'description: "{name} test"\n'
        "tools:\n"
        "  - name: builtin/read_file\n"
    )
    fd, path = tempfile.mkstemp(suffix=".yaml")
    with os.fdopen(fd, "w") as fh:
        fh.write(yaml_text)
    schema = SkillConfigSchema.from_yaml_file(path)
    Path(path).unlink(missing_ok=True)
    return schema


def cmd_persist_round_trip() -> None:
    """Add a skill, recreate service, verify visible."""
    factory, _engine = _setup()
    svc = _make_service(factory)

    config = _make_config("local/robot-persist")
    svc.add_skill(config, config_path="/tmp/robot.yaml")

    # Create a fresh service from the same DB
    svc2 = _make_service(factory)
    skills = svc2.list_skills()
    names = [s.name for s in skills]
    assert "local/robot-persist" in names, f"Expected skill in {names}"
    print("skill-add-persist-round-trip-ok")


def cmd_persist_multiple() -> None:
    """Add two skills, recreate service, verify both visible."""
    factory, _engine = _setup()
    svc = _make_service(factory)

    svc.add_skill(_make_config("local/multi-a"), config_path="/tmp/a.yaml")
    svc.add_skill(_make_config("local/multi-b"), config_path="/tmp/b.yaml")

    svc2 = _make_service(factory)
    skills = svc2.list_skills()
    names = [s.name for s in skills]
    assert "local/multi-a" in names, f"Expected multi-a in {names}"
    assert "local/multi-b" in names, f"Expected multi-b in {names}"
    assert len(names) == 2, f"Expected 2 skills, got {len(names)}"
    print("skill-add-persist-multiple-ok")


def cmd_persist_remove() -> None:
    """Add skill, remove, recreate service, verify gone."""
    factory, _engine = _setup()
    svc = _make_service(factory)

    svc.add_skill(_make_config("local/remove-me"), config_path="/tmp/rm.yaml")
    svc.remove_skill("local/remove-me")

    svc2 = _make_service(factory)
    skills = svc2.list_skills()
    names = [s.name for s in skills]
    assert "local/remove-me" not in names, f"Skill should be gone but found in {names}"
    print("skill-add-persist-remove-ok")


COMMANDS: dict[str, Any] = {
    "persist-round-trip": cmd_persist_round_trip,
    "persist-multiple": cmd_persist_multiple,
    "persist-remove": cmd_persist_remove,
}


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]]()