summaryrefslogtreecommitdiff
path: root/servers/hello_world/tests/test_integration.py
blob: 5948dae2db4dd8e4ba611018f7a5b59112848be9 (plain)
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
import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from io import StringIO
import json
from mcp_server_hello_world.server import HelloWorldServer, create_server

class TestIntegration:
    """Integration tests for the hello_world server."""
    
    @pytest.mark.asyncio
    async def test_hello_world_server(self):
        """Test the HelloWorldServer class."""
        # Create a HelloWorldServer instance
        server = HelloWorldServer()
        
        # Test the welcome message
        assert server.welcome_message == "Welcome to the Hello World MCP Server! This is a simple static resource."
        
        # Test the get_greeting method
        name = "Test User"
        expected_greeting = f"Hello, {name}! Welcome to the Hello World MCP Server."
        assert server.get_greeting(name) == expected_greeting
    
    @pytest.mark.asyncio
    async def test_server_initialization(self):
        """Test server initialization with stdio transport."""
        # Import the main function here to avoid issues with the mocking
        from mcp_server_hello_world.server import main
        
        # Mock the run_stdio_async method
        with patch('mcp.server.fastmcp.FastMCP.run_stdio_async') as mock_run_stdio:
            
            # Set up the mocks
            mock_run_stdio.return_value = AsyncMock()
            
            # Start the server with stdio transport
            task = asyncio.create_task(main("stdio", "localhost", 8080))
            
            # Wait for the server to start
            await asyncio.sleep(0.1)
            
            # Check that run_stdio_async was called
            mock_run_stdio.assert_called_once()
            
            # Cancel the task
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass
    
    @pytest.mark.asyncio
    async def test_server_functionality(self):
        """Test the server's functionality."""
        # Create a server
        mcp = create_server()
        
        # Test the resource
        resource = await mcp.read_resource("hello://welcome")
        assert len(resource) == 1
        assert resource[0].content == "Welcome to the Hello World MCP Server! This is a simple static resource."
        
        # Test the tool
        greeting = await mcp.call_tool("hello", {"name": "Test User"})
        assert len(greeting) == 1
        assert greeting[0].text == "Hello, Test User! Welcome to the Hello World MCP Server."
        
        # Test the prompt
        prompt = await mcp.get_prompt("greeting", {"name": "Test User"})
        assert len(prompt.messages) > 0
        assert prompt.messages[0].role == "user"
        assert prompt.messages[0].content.type == "text"
        assert "Hello there! You've chosen to use the greeting prompt with the name: Test User." in prompt.messages[0].content.text