98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
|
|
"""Tests for the Small Skill Registry (app/ai/skill_registry.py)."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app.ai.skill_registry import SkillDefinition, get_skill_registry
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture(autouse=True)
|
||
|
|
def _clean_registry():
|
||
|
|
"""Reset the singleton registry between tests."""
|
||
|
|
registry = get_skill_registry()
|
||
|
|
for skill in list(registry.list_all()):
|
||
|
|
registry.unregister(skill.name)
|
||
|
|
yield
|
||
|
|
for skill in list(registry.list_all()):
|
||
|
|
registry.unregister(skill.name)
|
||
|
|
|
||
|
|
|
||
|
|
def _make_skill(
|
||
|
|
name: str = "mail_reader",
|
||
|
|
allowed_tool_ids: list[str] | None = None,
|
||
|
|
category: str = "general",
|
||
|
|
) -> SkillDefinition:
|
||
|
|
return SkillDefinition(
|
||
|
|
name=name,
|
||
|
|
description="Read and summarize emails",
|
||
|
|
instructions="Use the mail tools to read and summarize emails.",
|
||
|
|
allowed_tool_ids=allowed_tool_ids or ["mail_read", "mail_list"],
|
||
|
|
context_policy={"include_recent": True},
|
||
|
|
category=category,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class TestSkillDefinition:
|
||
|
|
def test_defaults(self):
|
||
|
|
skill = SkillDefinition(
|
||
|
|
name="s", description="d", instructions="i", allowed_tool_ids=["t"]
|
||
|
|
)
|
||
|
|
assert skill.context_policy is None
|
||
|
|
assert skill.category == "general"
|
||
|
|
|
||
|
|
def test_to_dict(self):
|
||
|
|
skill = _make_skill()
|
||
|
|
data = skill.to_dict()
|
||
|
|
assert data["name"] == "mail_reader"
|
||
|
|
assert data["allowed_tool_ids"] == ["mail_read", "mail_list"]
|
||
|
|
assert data["context_policy"] == {"include_recent": True}
|
||
|
|
assert data["category"] == "general"
|
||
|
|
|
||
|
|
|
||
|
|
class TestSkillRegistry:
|
||
|
|
def test_register_and_get(self):
|
||
|
|
registry = get_skill_registry()
|
||
|
|
skill = _make_skill()
|
||
|
|
registry.register(skill)
|
||
|
|
assert registry.get("mail_reader") is skill
|
||
|
|
|
||
|
|
def test_get_missing_returns_none(self):
|
||
|
|
registry = get_skill_registry()
|
||
|
|
assert registry.get("nope") is None
|
||
|
|
|
||
|
|
def test_get_by_names_skips_unknown(self):
|
||
|
|
registry = get_skill_registry()
|
||
|
|
a = _make_skill("a")
|
||
|
|
b = _make_skill("b")
|
||
|
|
registry.register(a)
|
||
|
|
registry.register(b)
|
||
|
|
resolved = registry.get_by_names(["a", "b", "missing"])
|
||
|
|
assert len(resolved) == 2
|
||
|
|
assert {s.name for s in resolved} == {"a", "b"}
|
||
|
|
|
||
|
|
def test_list_all_and_list_for_api(self):
|
||
|
|
registry = get_skill_registry()
|
||
|
|
registry.register(_make_skill("a"))
|
||
|
|
registry.register(_make_skill("b"))
|
||
|
|
assert len(registry.list_all()) == 2
|
||
|
|
api = registry.list_for_api()
|
||
|
|
assert len(api) == 2
|
||
|
|
assert all("name" in item and "instructions" in item for item in api)
|
||
|
|
|
||
|
|
def test_unregister(self):
|
||
|
|
registry = get_skill_registry()
|
||
|
|
registry.register(_make_skill())
|
||
|
|
registry.unregister("mail_reader")
|
||
|
|
assert registry.get("mail_reader") is None
|
||
|
|
|
||
|
|
def test_register_replaces_existing(self):
|
||
|
|
registry = get_skill_registry()
|
||
|
|
registry.register(_make_skill("a", allowed_tool_ids=["t1"]))
|
||
|
|
registry.register(_make_skill("a", allowed_tool_ids=["t2"]))
|
||
|
|
assert registry.get("a").allowed_tool_ids == ["t2"]
|
||
|
|
assert len(registry.list_all()) == 1
|
||
|
|
|
||
|
|
def test_singleton(self):
|
||
|
|
assert get_skill_registry() is get_skill_registry()
|