molecule-core/workspace-template/tests/test_common_setup.py

"""Tests for the shared _common_setup() pipeline and tool conversion helpers."""

import importlib.util
import sys
from types import ModuleType
from unittest.mock import AsyncMock, MagicMock, patch

import pytest


# --- Mock missing optional deps ---

def _ensure_crewai_mock():
    if "crewai" not in sys.modules:
        crewai_mod = ModuleType("crewai")
        crewai_tools_mod = ModuleType("crewai.tools")
        # Make @tool a passthrough decorator that preserves the function
        crewai_tools_mod.tool = lambda name: (lambda f: f)
        crewai_mod.tools = crewai_tools_mod
        crewai_mod.__version__ = "0.0.0-mock"
        sys.modules["crewai"] = crewai_mod
        sys.modules["crewai.tools"] = crewai_tools_mod


def _ensure_autogen_mock():
    if "autogen_agentchat" not in sys.modules:
        mod = ModuleType("autogen_agentchat")
        agents_mod = ModuleType("autogen_agentchat.agents")
        agents_mod.AssistantAgent = MagicMock
        mod.agents = agents_mod
        sys.modules["autogen_agentchat"] = mod
        sys.modules["autogen_agentchat.agents"] = agents_mod


_ensure_crewai_mock()
_ensure_autogen_mock()


# --- Mock helpers ---

def _mock_load_plugins(**kwargs):
    plugins = MagicMock()
    plugins.plugin_names = []
    plugins.skill_dirs = []
    plugins.prompt_fragments = []
    plugins.rules = []
    return plugins


def _mock_load_skills(config_path, tools):
    return []


async def _mock_get_children():
    return []


async def _mock_get_children_with_kids():
    return [{"id": "child-1", "name": "Child", "role": "Worker", "status": "online"}]


async def _mock_get_parent_context():
    return []


async def _mock_get_peer_capabilities(platform_url, workspace_id):
    return [{"id": "peer-1", "name": "Peer", "status": "online", "agent_card": {"skills": []}}]


def _mock_build_system_prompt(*args, **kwargs):
    return "You are a test agent."


def _mock_build_children_description(children):
    return "## Team\n- Child: Worker"


# All patches needed for _common_setup
_SETUP_PATCHES = {
    "plugins.load_plugins": _mock_load_plugins,
    "skill_loader.loader.load_skills": _mock_load_skills,
    "coordinator.get_children": _mock_get_children,
    "coordinator.get_parent_context": _mock_get_parent_context,
    "coordinator.build_children_description": _mock_build_children_description,
    "prompt.get_peer_capabilities": _mock_get_peer_capabilities,
    "prompt.build_system_prompt": _mock_build_system_prompt,
}


def _make_test_adapter():
    from adapters.base import BaseAdapter, AdapterConfig

    class TestAdapter(BaseAdapter):
        @staticmethod
        def name(): return "test"
        @staticmethod
        def display_name(): return "Test"
        @staticmethod
        def description(): return "Test adapter"
        async def setup(self, config): pass
        async def create_executor(self, config): pass

    return TestAdapter(), AdapterConfig(model="openai:test", config_path="/tmp", workspace_id="ws-test")


# --- Common Setup Tests ---

@pytest.mark.asyncio
async def test_common_setup_returns_core_tools():
    """_common_setup returns 5 core tools."""
    adapter, config = _make_test_adapter()

    patches = {k: v for k, v in _SETUP_PATCHES.items()}
    with patch.dict("os.environ", {"PLATFORM_URL": "http://test:8080"}):
        ctx = [patch(k, v) for k, v in patches.items()]
        for c in ctx:
            c.start()
        try:
            result = await adapter._common_setup(config)
        finally:
            for c in ctx:
                c.stop()

    assert len(result.langchain_tools) == 6  # 6 core tools
    tool_names = [t.name for t in result.langchain_tools]
    assert "delegate_to_workspace" in tool_names
    assert "check_delegation_status" in tool_names
    assert "request_approval" in tool_names
    assert "commit_memory" in tool_names
    assert "search_memory" in tool_names
    assert "run_code" in tool_names
    assert result.system_prompt == "You are a test agent."
    assert result.is_coordinator is False


@pytest.mark.asyncio
async def test_common_setup_coordinator_adds_routing_tool():
    """When workspace has children, coordinator tool is added."""
    adapter, config = _make_test_adapter()

    patches = {k: v for k, v in _SETUP_PATCHES.items()}
    patches["coordinator.get_children"] = _mock_get_children_with_kids

    with patch.dict("os.environ", {"PLATFORM_URL": "http://test:8080"}):
        ctx = [patch(k, v) for k, v in patches.items()]
        for c in ctx:
            c.start()
        try:
            result = await adapter._common_setup(config)
        finally:
            for c in ctx:
                c.stop()

    assert result.is_coordinator is True
    assert len(result.langchain_tools) == 7  # 6 core + route_task_to_team
    # Last tool should be route_task_to_team (function name or .name attribute)
    last_tool = result.langchain_tools[-1]
    tool_id = getattr(last_tool, "name", None) or getattr(last_tool, "__name__", "")
    assert "route_task_to_team" in tool_id


# --- Tool Conversion Tests ---

def test_langchain_to_crewai_preserves_name():
    """CrewAI wrapper preserves tool name and description."""
    from adapters.crewai.adapter import _langchain_to_crewai

    mock_tool = MagicMock()
    mock_tool.name = "test_tool"
    mock_tool.description = "A test tool for testing."
    mock_tool.ainvoke = AsyncMock(return_value={"result": "ok"})

    wrapped = _langchain_to_crewai(mock_tool)
    # With our mock @tool decorator, the wrapper is the raw function
    assert wrapped.__doc__ == "A test tool for testing."


@pytest.mark.skipif(
    not importlib.util.find_spec("autogen_core"),
    reason="autogen_core not installed",
)
def test_langchain_to_autogen_preserves_name():
    """AutoGen wrapper preserves tool name and description via FunctionTool."""
    from adapters.autogen.adapter import _langchain_to_autogen

    mock_tool = MagicMock()
    mock_tool.name = "test_tool"
    mock_tool.description = "A test tool for testing."
    mock_tool.ainvoke = AsyncMock(return_value={"result": "ok"})

    wrapped = _langchain_to_autogen(mock_tool)
    assert wrapped.name == "test_tool"
    assert wrapped.description == "A test tool for testing."


@pytest.mark.skipif(
    not importlib.util.find_spec("autogen_core"),
    reason="autogen_core not installed",
)
@pytest.mark.asyncio
async def test_langchain_to_autogen_calls_ainvoke():
    """AutoGen FunctionTool wrapper calls the original tool's ainvoke."""
    from adapters.autogen.adapter import _langchain_to_autogen

    mock_tool = MagicMock()
    mock_tool.name = "delegate"
    mock_tool.description = "Delegate a task."
    mock_tool.ainvoke = AsyncMock(return_value={"success": True})

    wrapped = _langchain_to_autogen(mock_tool)
    # FunctionTool.run_json expects a JSON dict with the function params
    result = await wrapped.run_json({"input": '{"workspace_id": "ws-1", "task": "do stuff"}'}, cancellation_token=None)
    mock_tool.ainvoke.assert_called_once_with({"workspace_id": "ws-1", "task": "do stuff"})
    assert "True" in str(result)