mcp-forge/tests/server/test_resources.py

"""Tests for MCP resource handlers."""

import pytest
import json
from unittest.mock import Mock, AsyncMock, patch
from mcp_forge.server.resources import ResourceHandler


@pytest.fixture
def mock_client_manager():
    """Mock MCP client manager."""
    manager = Mock()
    manager.list_all_tools = AsyncMock(return_value=["read_file", "write_file", "calculate"])
    return manager


@pytest.fixture
def mock_session_manager():
    """Mock session manager."""
    manager = Mock()
    
    # Mock session state
    mock_state = Mock()
    mock_state.session_id = "test-session"
    mock_state.documented_variables = {"x": "Test variable", "result": "Calculation result"}
    mock_state.note = "Test session state"
    mock_state.all_variables = ["x", "y", "result", "np", "pd"]
    mock_state.introspection = {
        "x": {"type": "int", "size": 28},
        "result": {"type": "float", "size": 24}
    }
    mock_state.to_dict = Mock(return_value={
        "session_id": "test-session",
        "documented_variables": {"x": "Test variable", "result": "Calculation result"},
        "note": "Test session state",
        "all_variables": ["x", "y", "result", "np", "pd"],
        "introspection": {
            "x": {"type": "int", "size": 28},
            "result": {"type": "float", "size": 24}
        },
        "last_updated": "2026-02-06T12:00:00Z"
    })
    
    manager.get_session_state = Mock(return_value=mock_state)
    manager.list_sessions = Mock(return_value=[
        {"session_id": "test-session", "created_at": "2026-02-06T12:00:00Z"}
    ])
    
    return manager


@pytest.fixture
def mock_environment_builder():
    """Mock environment builder."""
    builder = Mock()
    builder.list_templates = Mock(return_value={
        "datascience": {
            "description": "Data science environment with numpy, pandas, matplotlib",
            "packages": ["numpy>=1.24", "pandas>=2.0", "matplotlib>=3.7"]
        },
        "ml": {
            "description": "Machine learning environment",
            "packages": ["scikit-learn>=1.3", "tensorflow>=2.13"]
        }
    })
    return builder


@pytest.fixture
def mock_config():
    """Mock forge configuration."""
    config = Mock()
    config.execution = Mock()
    config.execution.default_backend = "simple"
    config.execution.default_timeout = 300
    config.sessions = Mock()
    config.sessions.max_concurrent = 10
    # Add environment-related config for environment/info resource
    config.max_packages = 50
    config.max_build_time = 300
    config.base_images = {"python:3.11": {}, "python:3.12": {}}
    return config


@pytest.fixture
def resource_handler(mock_client_manager, mock_session_manager, mock_environment_builder, mock_config):
    """Create ResourceHandler instance with mocked dependencies."""
    return ResourceHandler(
        client_manager=mock_client_manager,
        session_manager=mock_session_manager,
        environment_builder=mock_environment_builder,
        config=mock_config
    )


@pytest.mark.asyncio
async def test_handle_tools_available(resource_handler, mock_client_manager):
    """Test handling tools/available resource returns list of available tools."""
    result = await resource_handler.handle_resource("mcp://forge/tools/available")
    
    assert str(result.uri) == "mcp://forge/tools/available"
    assert result.mimeType == "application/json"
    
    # Parse JSON content
    tools = json.loads(result.text)
    assert tools == {"tools": ["read_file", "write_file", "calculate"]}
    
    # Verify client manager was called
    mock_client_manager.list_all_tools.assert_awaited_once()


@pytest.mark.asyncio
async def test_handle_session_state(resource_handler, mock_session_manager):
    """Test handling session state resource returns documented state."""
    session_id = "test-session"
    result = await resource_handler.handle_resource(f"mcp://forge/sessions/{session_id}/state")
    
    assert str(result.uri) == f"mcp://forge/sessions/{session_id}/state"
    assert result.mimeType == "application/json"
    
    # Parse JSON content
    state = json.loads(result.text)
    assert state["session_id"] == session_id
    assert state["documented_variables"] == {"x": "Test variable", "result": "Calculation result"}
    assert state["note"] == "Test session state"
    assert "last_updated" in state
    
    # Verify session manager was called
    mock_session_manager.get_session_state.assert_called_once_with(session_id)


@pytest.mark.asyncio
async def test_handle_session_variables(resource_handler, mock_session_manager):
    """Test handling session variables resource returns list of variables."""
    session_id = "test-session"
    result = await resource_handler.handle_resource(f"mcp://forge/sessions/{session_id}/variables")
    
    assert str(result.uri) == f"mcp://forge/sessions/{session_id}/variables"
    assert result.mimeType == "application/json"
    
    # Parse JSON content
    data = json.loads(result.text)
    assert data["variables"] == ["x", "y", "result", "np", "pd"]
    
    # Verify session manager was called
    mock_session_manager.get_session_state.assert_called_once_with(session_id)


@pytest.mark.asyncio
async def test_handle_environments_list(resource_handler, mock_environment_builder):
    """Test handling environments/list resource returns templates and built environments."""
    result = await resource_handler.handle_resource("mcp://forge/environments/list")
    
    assert str(result.uri) == "mcp://forge/environments/list"
    assert result.mimeType == "application/json"
    
    # Parse JSON content
    environments = json.loads(result.text)
    assert "templates" in environments
    assert "datascience" in environments["templates"]
    assert "ml" in environments["templates"]
    assert environments["templates"]["datascience"]["description"] == "Data science environment with numpy, pandas, matplotlib"
    
    # Verify environment builder was called
    mock_environment_builder.list_templates.assert_called_once()


@pytest.mark.asyncio
async def test_handle_environment_info(resource_handler, mock_config):
    """Test handling environment info resource returns configuration info."""
    result = await resource_handler.handle_resource("mcp://forge/environment/info")
    
    assert str(result.uri) == "mcp://forge/environment/info"
    assert result.mimeType == "application/json"
    
    # Parse JSON content
    info = json.loads(result.text)
    assert "base_images" in info
    assert "python:3.11" in info["base_images"]
    assert "python:3.12" in info["base_images"]
    assert info["max_packages"] == 50
    assert info["max_build_time"] == 300


@pytest.mark.asyncio
async def test_handle_sessions_list(resource_handler, mock_session_manager):
    """Test handling sessions/list resource returns list of active sessions."""
    result = await resource_handler.handle_resource("mcp://forge/sessions/list")
    
    assert str(result.uri) == "mcp://forge/sessions/list"
    assert result.mimeType == "application/json"
    
    # Parse JSON content
    data = json.loads(result.text)
    assert len(data["sessions"]) == 1
    assert data["sessions"][0]["session_id"] == "test-session"
    
    # Verify session manager was called
    mock_session_manager.list_sessions.assert_called_once()


@pytest.mark.asyncio
async def test_handle_unknown_resource(resource_handler):
    """Test handling unknown resource raises ValueError."""
    with pytest.raises(ValueError, match="Unknown resource URI"):
        await resource_handler.handle_resource("mcp://forge/unknown/resource")


@pytest.mark.asyncio
async def test_handle_session_not_found(resource_handler, mock_session_manager):
    """Test handling session resource when session doesn't exist raises KeyError."""
    mock_session_manager.get_session_state.side_effect = KeyError("Session not found")
    
    with pytest.raises(KeyError, match="Session not found"):
        await resource_handler.handle_resource("mcp://forge/sessions/nonexistent/state")


@pytest.mark.asyncio
async def test_parse_uri_tools_available(resource_handler):
    """Test URI parsing for tools/available resource."""
    resource_type, params = resource_handler._parse_uri("mcp://forge/tools/available")
    assert resource_type == "tools_available"
    assert params == {}


@pytest.mark.asyncio
async def test_parse_uri_session_state(resource_handler):
    """Test URI parsing for session state resource."""
    resource_type, params = resource_handler._parse_uri("mcp://forge/sessions/test-123/state")
    assert resource_type == "session_state"
    assert params == {"session_id": "test-123"}


@pytest.mark.asyncio
async def test_parse_uri_session_variables(resource_handler):
    """Test URI parsing for session variables resource."""
    resource_type, params = resource_handler._parse_uri("mcp://forge/sessions/test-123/variables")
    assert resource_type == "session_variables"
    assert params == {"session_id": "test-123"}


@pytest.mark.asyncio
async def test_parse_uri_invalid_format(resource_handler):
    """Test URI parsing with invalid format raises ValueError."""
    with pytest.raises(ValueError, match="Invalid URI format"):
        resource_handler._parse_uri("invalid://uri")


@pytest.mark.asyncio
async def test_json_serialization_valid(resource_handler):
    """Test that all returned JSON content is valid and can be parsed."""
    # Test all resource types return valid JSON
    resources = [
        "mcp://forge/tools/available",
        "mcp://forge/sessions/test-session/state",
        "mcp://forge/sessions/test-session/variables",
        "mcp://forge/environments/list",
        "mcp://forge/environment/info",
        "mcp://forge/sessions/list"
    ]
    
    for uri in resources:
        result = await resource_handler.handle_resource(uri)
        # Should not raise exception
        parsed = json.loads(result.text)
        assert parsed is not None