tests/kg_services/test_ontology_mcp_enhancement.py

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