cleveragents-core/robot/helper_skill_registry_persistence.py

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

Usage:
    python helper_skill_registry_persistence.py <command>

Commands:
    persist-round-trip    Create a skill, persist flattened tools, retrieve
    update-invalidation   Update a skill and verify cache invalidation
"""

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

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
    SkillRepository,
)


def _enable_fk_pragma(dbapi_conn: Any, _connection_record: Any) -> None:
    """Enable SQLite foreign key enforcement per connection."""
    cursor = dbapi_conn.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()


def _setup() -> tuple[SkillRepository, sessionmaker[Session]]:
    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)
    return repo, factory


def _commit(factory: sessionmaker[Session]) -> None:
    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_persist_round_trip() -> None:
    """Create skill, persist flattened tools, retrieve and verify."""
    repo, factory = _setup()

    skill = _make_skill("local/persist-test", "Persistence test")
    repo.create(skill)
    _commit(factory)

    repo.update_flattened_tools(
        name="local/persist-test",
        flattened_tools_json='["tool-a","tool-b"]',
        includes_json="[]",
        capability_summary_json='{"total_tools": 2}',
        yaml_text="name: local/persist-test",
        flattening_hash="deadbeef",
    )
    _commit(factory)

    data = repo.get_flattened_tools("local/persist-test")
    assert data["flattened_tools_json"] == '["tool-a","tool-b"]'
    assert data["flattening_hash"] == "deadbeef"
    assert data["capability_summary_json"] == '{"total_tools": 2}'

    print("persist-round-trip-ok")


def cmd_update_invalidation() -> None:
    """Update skill and verify cached fields are nulled."""
    repo, factory = _setup()

    skill = _make_skill("local/invalidate-test", "Before update")
    repo.create(skill)
    _commit(factory)

    repo.update_flattened_tools(
        name="local/invalidate-test",
        flattened_tools_json='["x"]',
        includes_json="[]",
        capability_summary_json="{}",
        yaml_text="name: local/invalidate-test",
        flattening_hash="hash1",
    )
    _commit(factory)

    # Now update the skill — should null the cache
    updated = _make_skill("local/invalidate-test", "After update")
    repo.update(updated)
    _commit(factory)

    data = repo.get_flattened_tools("local/invalidate-test")
    assert data["flattened_tools_json"] is None, (
        f"Expected None, got {data['flattened_tools_json']}"
    )
    assert data["flattening_hash"] is None, (
        f"Expected None, got {data['flattening_hash']}"
    )

    print("update-invalidation-ok")


COMMANDS: dict[str, Any] = {
    "persist-round-trip": cmd_persist_round_trip,
    "update-invalidation": cmd_update_invalidation,
}


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