File size: 10,853 Bytes
1f2d50a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 |
"""Tests for enhanced MCPTool with MCP server support.
This module tests the enhanced ontology with new MCP server execution
fields added for MVP4 implementation.
"""
import pytest
from kg_services.ontology import MCPPrompt, MCPTool, PlannedStep
class TestMCPToolMCPEnhancement:
"""Test the enhanced MCPTool with MCP server support."""
def test_mcptool_default_execution_type(self):
"""Test that MCPTool defaults to simulated execution."""
tool = MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
)
assert tool.execution_type == "simulated"
assert tool.mcp_endpoint_url is None
assert tool.input_parameter_order == []
assert tool.timeout_seconds == 30
assert tool.requires_auth is False
def test_mcptool_simulated_execution(self):
"""Test MCPTool with simulated execution type."""
tool = MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
execution_type="simulated",
tags=["test"],
)
assert tool.execution_type == "simulated"
assert tool.tool_id == "test-tool"
assert tool.name == "Test Tool"
assert tool.description == "A test tool"
assert tool.tags == ["test"]
def test_mcptool_remote_mcp_execution(self):
"""Test MCPTool with remote MCP execution type."""
tool = MCPTool(
tool_id="summarizer-mcp",
name="Text Summarizer MCP",
description="Remote text summarizer via MCP",
execution_type="remote_mcp_gradio",
mcp_endpoint_url="https://example-summarizer.hf.space/mcp",
input_parameter_order=["text_content", "max_length"],
timeout_seconds=45,
requires_auth=True,
)
assert tool.execution_type == "remote_mcp_gradio"
assert tool.mcp_endpoint_url == "https://example-summarizer.hf.space/mcp"
assert tool.input_parameter_order == ["text_content", "max_length"]
assert tool.timeout_seconds == 45
assert tool.requires_auth is True
def test_mcptool_invalid_execution_type(self):
"""Test that invalid execution type raises ValueError."""
with pytest.raises(ValueError, match="execution_type must be"):
MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
execution_type="invalid_type",
)
def test_mcptool_remote_mcp_missing_url(self):
"""Test that remote MCP execution requires endpoint URL."""
with pytest.raises(ValueError, match="mcp_endpoint_url is required"):
MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
execution_type="remote_mcp_gradio",
# Missing mcp_endpoint_url
)
def test_mcptool_remote_mcp_invalid_url(self):
"""Test that MCP endpoint URL must be valid HTTP/HTTPS."""
with pytest.raises(ValueError, match="must be a valid HTTP/HTTPS URL"):
MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
execution_type="remote_mcp_gradio",
mcp_endpoint_url="invalid-url",
)
def test_mcptool_valid_http_url(self):
"""Test that HTTP URL is accepted for MCP endpoint."""
tool = MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
execution_type="remote_mcp_gradio",
mcp_endpoint_url="http://localhost:7860/mcp",
)
assert tool.mcp_endpoint_url == "http://localhost:7860/mcp"
def test_mcptool_valid_https_url(self):
"""Test that HTTPS URL is accepted for MCP endpoint."""
tool = MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
execution_type="remote_mcp_gradio",
mcp_endpoint_url="https://example.hf.space/mcp",
)
assert tool.mcp_endpoint_url == "https://example.hf.space/mcp"
def test_mcptool_negative_timeout(self):
"""Test that negative timeout raises ValueError."""
with pytest.raises(ValueError, match="timeout_seconds must be positive"):
MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
timeout_seconds=-1,
)
def test_mcptool_zero_timeout(self):
"""Test that zero timeout raises ValueError."""
with pytest.raises(ValueError, match="timeout_seconds must be positive"):
MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
timeout_seconds=0,
)
def test_mcptool_input_parameter_order(self):
"""Test that input parameter order is stored correctly."""
tool = MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
input_parameter_order=["param1", "param2", "param3"],
)
assert tool.input_parameter_order == ["param1", "param2", "param3"]
def test_mcptool_empty_input_parameter_order(self):
"""Test that empty input parameter order is allowed."""
tool = MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
input_parameter_order=[],
)
assert tool.input_parameter_order == []
def test_mcptool_with_all_fields(self):
"""Test MCPTool creation with all fields populated."""
tool = MCPTool(
tool_id="advanced-summarizer",
name="Advanced Text Summarizer",
description="AI-powered text summarization tool",
tags=["nlp", "summarization", "ai"],
invocation_command_stub="summarize --input {text} --max-len {length}",
execution_type="remote_mcp_gradio",
mcp_endpoint_url="https://my-summarizer.hf.space/mcp",
input_parameter_order=["text_content", "max_length", "min_length"],
timeout_seconds=60,
requires_auth=True,
)
# Verify all fields
assert tool.tool_id == "advanced-summarizer"
assert tool.name == "Advanced Text Summarizer"
assert tool.description == "AI-powered text summarization tool"
assert tool.tags == ["nlp", "summarization", "ai"]
assert tool.invocation_command_stub == "summarize --input {text} --max-len {length}"
assert tool.execution_type == "remote_mcp_gradio"
assert tool.mcp_endpoint_url == "https://my-summarizer.hf.space/mcp"
assert tool.input_parameter_order == ["text_content", "max_length", "min_length"]
assert tool.timeout_seconds == 60
assert tool.requires_auth is True
class TestPlannedStepWithMCPTool:
"""Test PlannedStep compatibility with enhanced MCPTool."""
def test_planned_step_with_mcp_tool(self):
"""Test that PlannedStep works with MCP-enhanced tools."""
tool = MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
execution_type="remote_mcp_gradio",
mcp_endpoint_url="https://example.hf.space/mcp",
)
prompt = MCPPrompt(
prompt_id="test-prompt",
name="Test Prompt",
description="A test prompt",
target_tool_id="test-tool",
template_string="Process: {{input}}",
input_variables=["input"],
)
step = PlannedStep(tool=tool, prompt=prompt, relevance_score=0.95)
assert step.tool.execution_type == "remote_mcp_gradio"
assert step.tool.mcp_endpoint_url == "https://example.hf.space/mcp"
assert step.relevance_score == 0.95
def test_planned_step_summary_with_mcp_tool(self):
"""Test PlannedStep summary with MCP tool."""
tool = MCPTool(
tool_id="mcp-summarizer",
name="MCP Summarizer",
description="Remote summarizer",
execution_type="remote_mcp_gradio",
mcp_endpoint_url="https://summarizer.hf.space/mcp",
)
prompt = MCPPrompt(
prompt_id="summary-prompt",
name="Detailed Summary",
description="Create detailed summary",
target_tool_id="mcp-summarizer",
template_string="Summarize: {{text}}",
)
step = PlannedStep(tool=tool, prompt=prompt)
assert step.summary == "Use 'MCP Summarizer' with 'Detailed Summary' prompt"
class TestMCPToolJSONCompatibility:
"""Test JSON serialization compatibility for enhanced MCPTool."""
def test_mcptool_dict_conversion(self):
"""Test that MCPTool can be converted to dict for JSON."""
from dataclasses import asdict
tool = MCPTool(
tool_id="test-tool",
name="Test Tool",
description="A test tool",
execution_type="remote_mcp_gradio",
mcp_endpoint_url="https://example.hf.space/mcp",
input_parameter_order=["param1", "param2"],
timeout_seconds=45,
requires_auth=True,
)
tool_dict = asdict(tool)
# Verify key fields in dict
assert tool_dict["tool_id"] == "test-tool"
assert tool_dict["execution_type"] == "remote_mcp_gradio"
assert tool_dict["mcp_endpoint_url"] == "https://example.hf.space/mcp"
assert tool_dict["input_parameter_order"] == ["param1", "param2"]
assert tool_dict["timeout_seconds"] == 45
assert tool_dict["requires_auth"] is True
def test_mcptool_from_dict(self):
"""Test MCPTool creation from dictionary (JSON loading)."""
tool_data = {
"tool_id": "json-tool",
"name": "JSON Tool",
"description": "Tool from JSON",
"execution_type": "remote_mcp_gradio",
"mcp_endpoint_url": "https://json-tool.hf.space/mcp",
"input_parameter_order": ["input1", "input2"],
"timeout_seconds": 30,
"requires_auth": False,
"tags": ["json", "test"],
"invocation_command_stub": "process {input}",
}
tool = MCPTool(**tool_data)
assert tool.tool_id == "json-tool"
assert tool.execution_type == "remote_mcp_gradio"
assert tool.mcp_endpoint_url == "https://json-tool.hf.space/mcp"
assert tool.input_parameter_order == ["input1", "input2"]
|