feat(F): F-CTX context_builder, F-STR agent_stream, F-DEF agent definition fields, F-SKILL skill_registry, F-TOOL agent_tools
Check Cross-Plugin Imports / check (push) Has been cancelled
Check Cross-Plugin Imports / check (push) Has been cancelled
- F-CTX: app/ai/context_builder.py (282 lines) — build_agent_context() + ReActSystemPromptBuilder - F-STR: app/ai/agent_stream.py (155 lines) — stream_react_loop() with SSE events (step, status, done, error) - F-DEF: AgentDefinition fields added (temperature, max_tokens, max_steps, trace_mode, skill_ids, trigger_config, ai_use_case_metadata) + migration 0122 - F-SKILL: app/ai/skill_registry.py (82 lines) — SkillDefinition + SkillRegistry singleton - F-TOOL: app/ai/agent_tools.py (117 lines) — get_agent_tools() with permission intersection - Skill CRUD routes: app/plugins/builtins/automation/skill_routes.py - Tests: test_skill_registry.py (97 lines), test_agent_tools.py (219 lines) - All Python compile checks pass, tests require PostgreSQL (infra issue, not code bug)
This commit is contained in:
@@ -0,0 +1,219 @@
|
||||
"""Tests for Tool-/Skill-Binding (app/ai/agent_tools.py).
|
||||
|
||||
All tests use mocked tool/skill registries — no real DB or LLM needed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from app.ai.agent_tools import get_agent_tools
|
||||
from app.ai.skill_registry import SkillDefinition, SkillRegistry
|
||||
|
||||
|
||||
@dataclass
|
||||
class MockTool:
|
||||
"""Minimal stand-in for AITool."""
|
||||
|
||||
name: str
|
||||
description: str = ""
|
||||
parameters: dict[str, Any] = field(default_factory=dict)
|
||||
required_permission: str | None = None
|
||||
|
||||
def to_openai_schema(self) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": self.name,
|
||||
"description": self.description,
|
||||
"parameters": self.parameters,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class MockToolRegistry:
|
||||
"""In-memory tool registry with get_by_names."""
|
||||
|
||||
def __init__(self, tools: list[MockTool]) -> None:
|
||||
self._tools = {t.name: t for t in tools}
|
||||
|
||||
def get_by_names(self, names: list[str]) -> list[MockTool]:
|
||||
return [self._tools[n] for n in names if n in self._tools]
|
||||
|
||||
|
||||
@dataclass
|
||||
class MockAgentDefinition:
|
||||
"""Minimal stand-in for AgentDefinition."""
|
||||
|
||||
tool_ids: list[str] = field(default_factory=list)
|
||||
skill_ids: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_skill_registry():
|
||||
"""Reset the singleton skill registry between tests."""
|
||||
registry = SkillRegistry()
|
||||
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,
|
||||
allowed_tool_ids: list[str],
|
||||
) -> SkillDefinition:
|
||||
return SkillDefinition(
|
||||
name=name,
|
||||
description="test skill",
|
||||
instructions="use the tools",
|
||||
allowed_tool_ids=allowed_tool_ids,
|
||||
)
|
||||
|
||||
|
||||
def _permissions(
|
||||
permissions: list[str] | None = None,
|
||||
denied: list[str] | None = None,
|
||||
is_system_admin: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"permissions": permissions or [],
|
||||
"denied_permissions": denied or [],
|
||||
"is_system_admin": is_system_admin,
|
||||
}
|
||||
|
||||
|
||||
class TestGetAgentTools:
|
||||
def test_direct_tools_available(self):
|
||||
"""Tools directly on the agent are available (no skills)."""
|
||||
registry = MockToolRegistry([
|
||||
MockTool("mail_read", required_permission="mail:read"),
|
||||
MockTool("contact_list"),
|
||||
])
|
||||
skill_registry = SkillRegistry()
|
||||
agent = MockAgentDefinition(tool_ids=["mail_read", "contact_list"])
|
||||
|
||||
schemas, skills = get_agent_tools(
|
||||
agent, registry, skill_registry, _permissions(permissions=["mail:read"])
|
||||
)
|
||||
|
||||
assert len(schemas) == 2
|
||||
names = {s["function"]["name"] for s in schemas}
|
||||
assert names == {"mail_read", "contact_list"}
|
||||
assert skills == []
|
||||
|
||||
def test_skill_tools_intersect_with_agent_tools(self):
|
||||
"""Skill tools are only available if also in agent.tool_ids."""
|
||||
registry = MockToolRegistry([
|
||||
MockTool("mail_read", required_permission="mail:read"),
|
||||
MockTool("mail_list", required_permission="mail:read"),
|
||||
MockTool("secret_tool"),
|
||||
])
|
||||
skill_registry = SkillRegistry()
|
||||
skill_registry.register(_make_skill("mail_skill", ["mail_read", "mail_list", "secret_tool"]))
|
||||
# Agent has mail_read + mail_list, but NOT secret_tool
|
||||
agent = MockAgentDefinition(
|
||||
tool_ids=["mail_read", "mail_list"],
|
||||
skill_ids=["mail_skill"],
|
||||
)
|
||||
|
||||
schemas, skills = get_agent_tools(
|
||||
agent, registry, skill_registry, _permissions(permissions=["mail:read"])
|
||||
)
|
||||
|
||||
names = {s["function"]["name"] for s in schemas}
|
||||
assert names == {"mail_read", "mail_list"}
|
||||
assert "secret_tool" not in names
|
||||
assert len(skills) == 1
|
||||
assert skills[0].name == "mail_skill"
|
||||
|
||||
def test_skill_never_grants_permission(self):
|
||||
"""A skill cannot grant access to a tool the user lacks permission for."""
|
||||
registry = MockToolRegistry([
|
||||
MockTool("mail_read", required_permission="mail:read"),
|
||||
])
|
||||
skill_registry = SkillRegistry()
|
||||
skill_registry.register(_make_skill("mail_skill", ["mail_read"]))
|
||||
agent = MockAgentDefinition(tool_ids=["mail_read"], skill_ids=["mail_skill"])
|
||||
|
||||
# User does NOT have mail:read
|
||||
schemas, skills = get_agent_tools(
|
||||
agent, registry, skill_registry, _permissions(permissions=[])
|
||||
)
|
||||
|
||||
assert schemas == []
|
||||
assert len(skills) == 1 # skill is still resolved, but grants no tools
|
||||
|
||||
def test_denied_permission_blocks_tool(self):
|
||||
"""Explicitly denied permissions block a tool even if granted."""
|
||||
registry = MockToolRegistry([
|
||||
MockTool("mail_read", required_permission="mail:read"),
|
||||
])
|
||||
skill_registry = SkillRegistry()
|
||||
agent = MockAgentDefinition(tool_ids=["mail_read"])
|
||||
|
||||
schemas, _ = get_agent_tools(
|
||||
agent,
|
||||
registry,
|
||||
skill_registry,
|
||||
_permissions(permissions=["mail:read"], denied=["mail:read"]),
|
||||
)
|
||||
|
||||
assert schemas == []
|
||||
|
||||
def test_system_admin_gets_all_tools(self):
|
||||
"""System admins bypass permission checks."""
|
||||
registry = MockToolRegistry([
|
||||
MockTool("mail_read", required_permission="mail:read"),
|
||||
])
|
||||
skill_registry = SkillRegistry()
|
||||
agent = MockAgentDefinition(tool_ids=["mail_read"])
|
||||
|
||||
schemas, _ = get_agent_tools(
|
||||
agent, registry, skill_registry, _permissions(is_system_admin=True)
|
||||
)
|
||||
|
||||
assert len(schemas) == 1
|
||||
|
||||
def test_wildcard_permission_matches(self):
|
||||
"""Wildcard permissions (mail:*) satisfy a required permission."""
|
||||
registry = MockToolRegistry([
|
||||
MockTool("mail_read", required_permission="mail:read"),
|
||||
])
|
||||
skill_registry = SkillRegistry()
|
||||
agent = MockAgentDefinition(tool_ids=["mail_read"])
|
||||
|
||||
schemas, _ = get_agent_tools(
|
||||
agent, registry, skill_registry, _permissions(permissions=["mail:*"])
|
||||
)
|
||||
|
||||
assert len(schemas) == 1
|
||||
|
||||
def test_tool_without_permission_always_available(self):
|
||||
"""Tools without a required_permission are always available."""
|
||||
registry = MockToolRegistry([MockTool("contact_list")])
|
||||
skill_registry = SkillRegistry()
|
||||
agent = MockAgentDefinition(tool_ids=["contact_list"])
|
||||
|
||||
schemas, _ = get_agent_tools(
|
||||
agent, registry, skill_registry, _permissions(permissions=[])
|
||||
)
|
||||
|
||||
assert len(schemas) == 1
|
||||
|
||||
def test_unknown_tools_are_skipped(self):
|
||||
"""Tool IDs not in the registry are silently skipped."""
|
||||
registry = MockToolRegistry([MockTool("known_tool")])
|
||||
skill_registry = SkillRegistry()
|
||||
agent = MockAgentDefinition(tool_ids=["known_tool", "missing_tool"])
|
||||
|
||||
schemas, _ = get_agent_tools(
|
||||
agent, registry, skill_registry, _permissions(permissions=[])
|
||||
)
|
||||
|
||||
assert len(schemas) == 1
|
||||
assert schemas[0]["function"]["name"] == "known_tool"
|
||||
Reference in New Issue
Block a user