mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-07-14 16:36:07 +00:00
@@ -280,6 +280,25 @@ class TestMCPServiceCreateMCPServer:
|
||||
assert server_uuid is not None
|
||||
assert len(server_uuid) == 36 # UUID format
|
||||
|
||||
async def test_create_mcp_server_duplicate_name_raises(self):
|
||||
"""Rejects duplicate MCP server names."""
|
||||
# Setup
|
||||
ap = SimpleNamespace()
|
||||
ap.persistence_mgr = SimpleNamespace()
|
||||
ap.instance_config = SimpleNamespace()
|
||||
ap.instance_config.data = {'system': {'limitation': {'max_extensions': -1}}}
|
||||
ap.tool_mgr = None
|
||||
|
||||
existing_server = _create_mock_mcp_server(name='Existing Server')
|
||||
ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_mock_result(first_item=existing_server))
|
||||
ap.persistence_mgr.serialize_model = Mock(return_value={})
|
||||
|
||||
service = MCPService(ap)
|
||||
|
||||
# Execute & Verify
|
||||
with pytest.raises(ValueError, match='MCP server already exists: Existing Server'):
|
||||
await service.create_mcp_server({'name': 'Existing Server'})
|
||||
|
||||
async def test_create_mcp_server_loads_server(self):
|
||||
"""Loads server into tool_mgr when enabled."""
|
||||
# Setup
|
||||
@@ -301,7 +320,7 @@ class TestMCPServiceCreateMCPServer:
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count == 1:
|
||||
return _create_mock_result([]) # Empty list for limit check
|
||||
return _create_mock_result([]) # Empty result for duplicate-name check
|
||||
elif call_count == 2:
|
||||
return Mock() # Insert
|
||||
return _create_mock_result(first_item=server_entity) # Select created
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import types
|
||||
from importlib import import_module
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
import quart
|
||||
|
||||
core_app_module = types.ModuleType('langbot.pkg.core.app')
|
||||
core_app_module.Application = object
|
||||
sys.modules.setdefault('langbot.pkg.core.app', core_app_module)
|
||||
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
async def _create_test_client(mcp_service: SimpleNamespace):
|
||||
app = quart.Quart(__name__)
|
||||
user_service = SimpleNamespace(
|
||||
verify_jwt_token=AsyncMock(return_value='test@example.com'),
|
||||
get_user_by_email=AsyncMock(return_value=SimpleNamespace(user='test@example.com')),
|
||||
)
|
||||
ap = SimpleNamespace(mcp_service=mcp_service, user_service=user_service)
|
||||
MCPRouterGroup = import_module('langbot.pkg.api.http.controller.groups.resources.mcp').MCPRouterGroup
|
||||
group = MCPRouterGroup(ap, app)
|
||||
await group.initialize()
|
||||
return app.test_client()
|
||||
|
||||
|
||||
async def test_mcp_server_route_accepts_encoded_slash_name():
|
||||
mcp_service = SimpleNamespace(
|
||||
get_mcp_server_by_name=AsyncMock(
|
||||
return_value={
|
||||
'uuid': 'test-uuid',
|
||||
'name': 'pab1it0/prometheus',
|
||||
'enable': True,
|
||||
'mode': 'stdio',
|
||||
'extra_args': {},
|
||||
}
|
||||
)
|
||||
)
|
||||
client = await _create_test_client(mcp_service)
|
||||
|
||||
response = await client.get(
|
||||
'/api/v1/mcp/servers/pab1it0%2Fprometheus',
|
||||
headers={'Authorization': 'Bearer test-token'},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
mcp_service.get_mcp_server_by_name.assert_awaited_once_with('pab1it0/prometheus')
|
||||
payload = await response.get_json()
|
||||
assert payload['data']['server']['name'] == 'pab1it0/prometheus'
|
||||
|
||||
|
||||
async def test_mcp_resource_route_accepts_encoded_slash_name():
|
||||
mcp_service = SimpleNamespace(
|
||||
get_mcp_server_by_name=AsyncMock(),
|
||||
get_mcp_server_resources=AsyncMock(return_value=[]),
|
||||
get_mcp_server_resource_templates=AsyncMock(return_value=[]),
|
||||
get_runtime_info=AsyncMock(return_value={'resource_capabilities': {'subscribe': False}}),
|
||||
)
|
||||
client = await _create_test_client(mcp_service)
|
||||
|
||||
response = await client.get(
|
||||
'/api/v1/mcp/servers/pab1it0%2Fprometheus/resources',
|
||||
headers={'Authorization': 'Bearer test-token'},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
mcp_service.get_mcp_server_by_name.assert_not_awaited()
|
||||
mcp_service.get_mcp_server_resources.assert_awaited_once_with('pab1it0/prometheus')
|
||||
payload = await response.get_json()
|
||||
assert payload['data']['resource_capabilities'] == {'subscribe': False}
|
||||
Reference in New Issue
Block a user