leocrm/tests/test_agent_tools.py

"""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"