mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-06-03 20:44:36 +00:00
d1ddff9cdb
The merge from master brought in new unit tests that target pre-refactor
APIs on feat/sandbox. Reconcile each:
- factories/app.py: FakeApp now exposes a Mock skill_mgr (with empty .skills
dict + inert prompt-addition builder) and a Mock pipeline_service so the
PreProcessor skill-index injection branch can run end-to-end in tests.
- pipeline/conftest.py: eagerly import langbot.pkg.pipeline.pipelinemgr so
pipeline.stage is fully initialised before any individual stage test
(preproc, longtext, ...) tries to lazy-load it. Without this preload,
running test_preproc.py in isolation hit a circular-import error via the
stage -> app -> pipelinemgr -> stage chain.
- provider/test_tool_manager.py: ToolManager now probes four loaders
(native -> plugin -> mcp -> skill). Inject inert native + skill mocks in
the execute_func_call fixture and assert all four shutdowns fire.
- utils/test_paths.py: drop the three cwd-dependent _check_if_source_install
cases. The refactor walks Path(__file__).resolve().parents looking for
pyproject.toml + main.py, so cwd no longer factors in and there's no
file read to mock-fail. The positive case and caching test still apply.
- utils/test_version.py: delete entirely. is_newer and compare_version_str
were removed when VersionManager was refactored to use the Space API for
release checks (1b4107a9); the tests targeted a surface that no longer
exists.
333 lines
12 KiB
Python
333 lines
12 KiB
Python
"""Unit tests for ToolManager.
|
|
|
|
Tests cover:
|
|
- Tool schema generation for OpenAI and Anthropic
|
|
- Tool execution dispatch
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, AsyncMock
|
|
from importlib import import_module
|
|
|
|
import langbot_plugin.api.entities.builtin.resource.tool as resource_tool
|
|
import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
|
|
|
|
|
|
def get_toolmgr_module():
|
|
"""Lazy import to avoid circular import issues."""
|
|
return import_module('langbot.pkg.provider.tools.toolmgr')
|
|
|
|
|
|
class TestToolManagerInit:
|
|
"""Tests for ToolManager initialization."""
|
|
|
|
def test_init_stores_app_reference(self):
|
|
"""Test that __init__ stores the Application reference."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
mock_app = Mock()
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
assert manager.ap is mock_app
|
|
|
|
def test_init_no_tool_loaders(self):
|
|
"""Test that tool loaders are not initialized before initialize()."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
mock_app = Mock()
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
assert hasattr(manager, 'plugin_tool_loader') is False or manager.plugin_tool_loader is None
|
|
|
|
|
|
class TestToolManagerSchemaGeneration:
|
|
"""Tests for tool schema generation methods."""
|
|
|
|
@pytest.fixture
|
|
def mock_app(self):
|
|
"""Create mock app."""
|
|
mock_app = Mock()
|
|
mock_app.logger = Mock()
|
|
return mock_app
|
|
|
|
@pytest.fixture
|
|
def sample_tools(self):
|
|
"""Create sample LLMTool list for testing."""
|
|
|
|
def dummy_weather_func(**kwargs):
|
|
return 'weather result'
|
|
|
|
def dummy_calc_func(**kwargs):
|
|
return 'calc result'
|
|
|
|
tools = [
|
|
resource_tool.LLMTool(
|
|
name='get_weather',
|
|
human_desc='Get current weather for a location',
|
|
description='Get current weather for a location',
|
|
parameters={
|
|
'type': 'object',
|
|
'properties': {'location': {'type': 'string', 'description': 'City name'}},
|
|
'required': ['location'],
|
|
},
|
|
func=dummy_weather_func,
|
|
),
|
|
resource_tool.LLMTool(
|
|
name='calculate',
|
|
human_desc='Perform a calculation',
|
|
description='Perform a calculation',
|
|
parameters={
|
|
'type': 'object',
|
|
'properties': {'expression': {'type': 'string', 'description': 'Math expression'}},
|
|
'required': ['expression'],
|
|
},
|
|
func=dummy_calc_func,
|
|
),
|
|
]
|
|
return tools
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_tools_for_openai(self, mock_app, sample_tools):
|
|
"""Test that generate_tools_for_openai produces correct schema."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
result = await manager.generate_tools_for_openai(sample_tools)
|
|
|
|
assert len(result) == 2
|
|
|
|
# Verify first tool schema
|
|
tool1 = result[0]
|
|
assert tool1['type'] == 'function'
|
|
assert tool1['function']['name'] == 'get_weather'
|
|
assert tool1['function']['description'] == 'Get current weather for a location'
|
|
assert 'parameters' in tool1['function']
|
|
assert tool1['function']['parameters']['type'] == 'object'
|
|
|
|
# Verify second tool schema
|
|
tool2 = result[1]
|
|
assert tool2['type'] == 'function'
|
|
assert tool2['function']['name'] == 'calculate'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_tools_for_anthropic(self, mock_app, sample_tools):
|
|
"""Test that generate_tools_for_anthropic produces correct schema."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
result = await manager.generate_tools_for_anthropic(sample_tools)
|
|
|
|
assert len(result) == 2
|
|
|
|
# Verify first tool schema (Anthropic format)
|
|
tool1 = result[0]
|
|
assert tool1['name'] == 'get_weather'
|
|
assert tool1['description'] == 'Get current weather for a location'
|
|
assert 'input_schema' in tool1
|
|
assert tool1['input_schema']['type'] == 'object'
|
|
|
|
# Verify second tool schema
|
|
tool2 = result[1]
|
|
assert tool2['name'] == 'calculate'
|
|
assert 'input_schema' in tool2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_tools_empty_list(self, mock_app):
|
|
"""Test that generating tools from empty list returns empty list."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
|
|
openai_result = await manager.generate_tools_for_openai([])
|
|
assert openai_result == []
|
|
|
|
anthropic_result = await manager.generate_tools_for_anthropic([])
|
|
assert anthropic_result == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_openai_schema_fields_complete(self, mock_app, sample_tools):
|
|
"""Test that OpenAI schema includes all required fields."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
result = await manager.generate_tools_for_openai(sample_tools)
|
|
|
|
for tool_schema in result:
|
|
assert 'type' in tool_schema
|
|
assert tool_schema['type'] == 'function'
|
|
assert 'function' in tool_schema
|
|
func = tool_schema['function']
|
|
assert 'name' in func
|
|
assert 'description' in func
|
|
assert 'parameters' in func
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anthropic_schema_fields_complete(self, mock_app, sample_tools):
|
|
"""Test that Anthropic schema includes all required fields."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
result = await manager.generate_tools_for_anthropic(sample_tools)
|
|
|
|
for tool_schema in result:
|
|
assert 'name' in tool_schema
|
|
assert 'description' in tool_schema
|
|
assert 'input_schema' in tool_schema
|
|
|
|
|
|
class TestToolManagerExecuteFuncCall:
|
|
"""Tests for execute_func_call method."""
|
|
|
|
@pytest.fixture
|
|
def mock_app_with_loaders(self):
|
|
"""Create mock app with mock tool loaders.
|
|
|
|
Returns (app, plugin_loader, mcp_loader). The native and skill loaders
|
|
are attached directly to the app for tests that don't need to assert
|
|
against them — they all default to ``has_tool == False`` so the
|
|
execute_func_call probe falls through to the plugin/mcp pair.
|
|
"""
|
|
mock_app = Mock()
|
|
mock_app.logger = Mock()
|
|
|
|
def _make_inert_loader():
|
|
loader = Mock()
|
|
loader.has_tool = AsyncMock(return_value=False)
|
|
loader.invoke_tool = AsyncMock(return_value=None)
|
|
loader.initialize = AsyncMock()
|
|
loader.shutdown = AsyncMock()
|
|
return loader
|
|
|
|
# Create mock plugin loader
|
|
mock_plugin_loader = _make_inert_loader()
|
|
mock_plugin_loader.invoke_tool = AsyncMock(return_value='plugin_result')
|
|
|
|
# Create mock MCP loader
|
|
mock_mcp_loader = _make_inert_loader()
|
|
mock_mcp_loader.invoke_tool = AsyncMock(return_value='mcp_result')
|
|
|
|
# Stash inert native/skill loaders so the ToolManager probe order
|
|
# (native → plugin → mcp → skill) doesn't AttributeError. Tests that
|
|
# need to override these can replace the attributes on the manager.
|
|
mock_app._inert_native_loader = _make_inert_loader()
|
|
mock_app._inert_skill_loader = _make_inert_loader()
|
|
|
|
return mock_app, mock_plugin_loader, mock_mcp_loader
|
|
|
|
@staticmethod
|
|
def _wire_loaders(manager, mock_app, plugin_loader, mcp_loader):
|
|
"""Attach all four loaders (native + plugin + mcp + skill) to manager."""
|
|
manager.native_tool_loader = mock_app._inert_native_loader
|
|
manager.plugin_tool_loader = plugin_loader
|
|
manager.mcp_tool_loader = mcp_loader
|
|
manager.skill_tool_loader = mock_app._inert_skill_loader
|
|
|
|
@pytest.fixture
|
|
def sample_query(self):
|
|
"""Create sample query for testing."""
|
|
query = Mock(spec=pipeline_query.Query)
|
|
return query
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_calls_plugin_loader_when_has_tool(self, mock_app_with_loaders, sample_query):
|
|
"""Test that execute_func_call uses plugin loader when tool exists there."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
mock_app, mock_plugin_loader, mock_mcp_loader = mock_app_with_loaders
|
|
mock_plugin_loader.has_tool = AsyncMock(return_value=True)
|
|
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
self._wire_loaders(manager, mock_app, mock_plugin_loader, mock_mcp_loader)
|
|
|
|
result = await manager.execute_func_call('test_tool', {'param': 'value'}, sample_query)
|
|
|
|
assert result == 'plugin_result'
|
|
mock_plugin_loader.invoke_tool.assert_called_once_with('test_tool', {'param': 'value'}, sample_query)
|
|
# MCP loader should not be called
|
|
mock_mcp_loader.invoke_tool.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_calls_mcp_loader_when_plugin_not_found(self, mock_app_with_loaders, sample_query):
|
|
"""Test that execute_func_call uses MCP loader when plugin doesn't have tool."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
mock_app, mock_plugin_loader, mock_mcp_loader = mock_app_with_loaders
|
|
mock_plugin_loader.has_tool = AsyncMock(return_value=False)
|
|
mock_mcp_loader.has_tool = AsyncMock(return_value=True)
|
|
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
self._wire_loaders(manager, mock_app, mock_plugin_loader, mock_mcp_loader)
|
|
|
|
result = await manager.execute_func_call('test_tool', {'param': 'value'}, sample_query)
|
|
|
|
assert result == 'mcp_result'
|
|
mock_mcp_loader.invoke_tool.assert_called_once_with('test_tool', {'param': 'value'}, sample_query)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_raises_when_tool_not_found(self, mock_app_with_loaders, sample_query):
|
|
"""Test that execute_func_call raises ValueError when tool not found."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
mock_app, mock_plugin_loader, mock_mcp_loader = mock_app_with_loaders
|
|
mock_plugin_loader.has_tool = AsyncMock(return_value=False)
|
|
mock_mcp_loader.has_tool = AsyncMock(return_value=False)
|
|
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
self._wire_loaders(manager, mock_app, mock_plugin_loader, mock_mcp_loader)
|
|
|
|
with pytest.raises(ValueError, match='未找到工具'):
|
|
await manager.execute_func_call('unknown_tool', {}, sample_query)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plugin_loader_checked_first(self, mock_app_with_loaders, sample_query):
|
|
"""Test that plugin loader is checked before MCP loader."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
mock_app, mock_plugin_loader, mock_mcp_loader = mock_app_with_loaders
|
|
# Both loaders have the tool, but plugin should be used
|
|
mock_plugin_loader.has_tool = AsyncMock(return_value=True)
|
|
mock_mcp_loader.has_tool = AsyncMock(return_value=True)
|
|
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
self._wire_loaders(manager, mock_app, mock_plugin_loader, mock_mcp_loader)
|
|
|
|
await manager.execute_func_call('test_tool', {}, sample_query)
|
|
|
|
# Plugin loader should be invoked, MCP should not
|
|
mock_plugin_loader.invoke_tool.assert_called_once()
|
|
mock_mcp_loader.invoke_tool.assert_not_called()
|
|
|
|
|
|
class TestToolManagerShutdown:
|
|
"""Tests for shutdown method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shutdown_calls_loader_shutdown(self):
|
|
"""Test that shutdown calls shutdown on every registered loader."""
|
|
toolmgr = get_toolmgr_module()
|
|
|
|
mock_app = Mock()
|
|
|
|
def _make_loader():
|
|
loader = Mock()
|
|
loader.shutdown = AsyncMock()
|
|
return loader
|
|
|
|
mock_native_loader = _make_loader()
|
|
mock_plugin_loader = _make_loader()
|
|
mock_mcp_loader = _make_loader()
|
|
mock_skill_loader = _make_loader()
|
|
|
|
manager = toolmgr.ToolManager(mock_app)
|
|
manager.native_tool_loader = mock_native_loader
|
|
manager.plugin_tool_loader = mock_plugin_loader
|
|
manager.mcp_tool_loader = mock_mcp_loader
|
|
manager.skill_tool_loader = mock_skill_loader
|
|
|
|
await manager.shutdown()
|
|
|
|
mock_native_loader.shutdown.assert_called_once()
|
|
mock_plugin_loader.shutdown.assert_called_once()
|
|
mock_mcp_loader.shutdown.assert_called_once()
|
|
mock_skill_loader.shutdown.assert_called_once()
|