244 lines
8.4 KiB
Python
244 lines
8.4 KiB
Python
"""Tests for Skill Management API routes"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from agentkit.llm.gateway import LLMGateway
|
|
from agentkit.server.app import create_app
|
|
from agentkit.skills.base import Skill, SkillConfig
|
|
from agentkit.skills.registry import SkillRegistry
|
|
from agentkit.tools.registry import ToolRegistry
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_llm_gateway():
|
|
return LLMGateway()
|
|
|
|
|
|
@pytest.fixture
|
|
def skill_registry():
|
|
return SkillRegistry()
|
|
|
|
|
|
@pytest.fixture
|
|
def tool_registry():
|
|
return ToolRegistry()
|
|
|
|
|
|
@pytest.fixture
|
|
def app(mock_llm_gateway, skill_registry, tool_registry):
|
|
return create_app(
|
|
llm_gateway=mock_llm_gateway,
|
|
skill_registry=skill_registry,
|
|
tool_registry=tool_registry,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
return TestClient(app)
|
|
|
|
|
|
def _register_skill(registry: SkillRegistry, name: str = "test_skill", **kwargs):
|
|
"""Helper to register a skill with sensible defaults."""
|
|
config = SkillConfig(
|
|
name=name,
|
|
agent_type="test_type",
|
|
task_mode="llm_generate",
|
|
prompt={"identity": "Test Skill", "instructions": "Handle test"},
|
|
intent={"keywords": ["test"], "description": "A test skill"},
|
|
**kwargs,
|
|
)
|
|
skill = Skill(config=config)
|
|
registry.register(skill)
|
|
return skill
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /skill-management/skills
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListSkills:
|
|
def test_list_skills_empty(self, client):
|
|
response = client.get("/api/v1/skill-management/skills")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "skills" in data
|
|
assert isinstance(data["skills"], list)
|
|
assert "total" in data
|
|
assert "page" in data
|
|
assert "size" in data
|
|
|
|
def test_list_skills_with_registered(self, client, skill_registry):
|
|
_register_skill(skill_registry, "skill_a")
|
|
_register_skill(skill_registry, "skill_b")
|
|
|
|
response = client.get("/api/v1/skill-management/skills")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] >= 2
|
|
assert len(data["skills"]) >= 2
|
|
|
|
def test_list_skills_pagination(self, client, skill_registry):
|
|
for i in range(5):
|
|
_register_skill(skill_registry, f"page_skill_{i}")
|
|
|
|
response = client.get("/api/v1/skill-management/skills?page=1&size=2")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["size"] == 2
|
|
assert len(data["skills"]) <= 2
|
|
assert data["total"] >= 5
|
|
|
|
def test_list_skills_skill_structure(self, client, skill_registry):
|
|
_register_skill(skill_registry, "struct_skill")
|
|
|
|
response = client.get("/api/v1/skill-management/skills")
|
|
data = response.json()
|
|
if data["skills"]:
|
|
skill = data["skills"][0]
|
|
assert "name" in skill
|
|
assert "version" in skill
|
|
assert "description" in skill
|
|
assert "capabilities" in skill
|
|
assert "dependencies" in skill
|
|
assert "status" in skill
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /skill-management/skills/{skill_name}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetSkillDetail:
|
|
def test_get_skill_detail(self, client, skill_registry):
|
|
_register_skill(skill_registry, "detail_skill")
|
|
|
|
response = client.get("/api/v1/skill-management/skills/detail_skill")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["name"] == "detail_skill"
|
|
assert "version" in data
|
|
assert "description" in data
|
|
assert "capabilities" in data
|
|
assert "dependencies" in data
|
|
assert "config" in data
|
|
assert "health_status" in data
|
|
|
|
def test_get_skill_detail_not_found(self, client):
|
|
response = client.get("/api/v1/skill-management/skills/nonexistent")
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /skill-management/skills/{skill_name}/health
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSkillHealth:
|
|
def test_skill_health(self, client, skill_registry):
|
|
_register_skill(skill_registry, "health_skill")
|
|
|
|
response = client.get("/api/v1/skill-management/skills/health_skill/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["skill_name"] == "health_skill"
|
|
assert data["status"] == "healthy"
|
|
|
|
def test_skill_health_not_found(self, client):
|
|
response = client.get("/api/v1/skill-management/skills/nonexistent/health")
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /skill-management/capabilities
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListCapabilities:
|
|
def test_list_capabilities_empty(self, client):
|
|
response = client.get("/api/v1/skill-management/capabilities")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "capabilities" in data
|
|
assert isinstance(data["capabilities"], list)
|
|
|
|
def test_list_capabilities_with_skills(self, client, skill_registry):
|
|
_register_skill(skill_registry, "cap_skill", capabilities=["chat"])
|
|
|
|
response = client.get("/api/v1/skill-management/capabilities")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["capabilities"]) >= 1
|
|
# Each capability should have name, display_name, skill_count
|
|
for cap in data["capabilities"]:
|
|
assert "name" in cap
|
|
assert "display_name" in cap
|
|
assert "skill_count" in cap
|
|
|
|
def test_list_capabilities_structure(self, client, skill_registry):
|
|
_register_skill(skill_registry, "multi_cap_skill", capabilities=["chat", "search"])
|
|
|
|
response = client.get("/api/v1/skill-management/capabilities")
|
|
data = response.json()
|
|
cap_names = [c["name"] for c in data["capabilities"]]
|
|
assert "chat" in cap_names
|
|
assert "search" in cap_names
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /skill-management/skills/{skill_name}/reload
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestReloadSkill:
|
|
def test_reload_skill(self, client, skill_registry):
|
|
_register_skill(skill_registry, "reload_skill")
|
|
|
|
response = client.post("/api/v1/skill-management/skills/reload_skill/reload")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["skill_name"] == "reload_skill"
|
|
assert data["status"] == "reloaded"
|
|
|
|
def test_reload_skill_not_found(self, client):
|
|
response = client.post("/api/v1/skill-management/skills/nonexistent/reload")
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Capability filtering
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSkillCapabilityFilter:
|
|
def test_filter_by_capability(self, client, skill_registry):
|
|
_register_skill(skill_registry, "chat_only_skill", capabilities=["chat"])
|
|
_register_skill(skill_registry, "search_only_skill", capabilities=["search"])
|
|
|
|
response = client.get("/api/v1/skill-management/skills?capability=chat")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
# All returned skills should have "chat" capability
|
|
for skill in data["skills"]:
|
|
assert "chat" in skill["capabilities"]
|
|
|
|
def test_filter_by_nonexistent_capability(self, client, skill_registry):
|
|
_register_skill(skill_registry, "some_skill")
|
|
|
|
response = client.get("/api/v1/skill-management/skills?capability=nonexistent_cap")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 0
|
|
assert len(data["skills"]) == 0
|